Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

OpenTracing for Fission (#1079)

Browse Source 
Added Opentracing integration using opencensus libraries for all Fission components.
This commit is contained in:
Vishal
2019-02-07 11:56:41 +05:30
committed by GitHub
parent 8b0a201f69
commit 5d2abdd95b
29 changed files with 400 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
buildermgr/common.go
+4 -3
View File
@@ -17,6 +17,7 @@ limitations under the License.
package buildermgr
import (
"context"
"fmt"
"log"
"net/http"
@@ -37,7 +38,7 @@ import (
// 3. Send upload request to fetcher to upload deployment package.
// 4. Return upload response and build logs.
// *. Return build logs and error if any one of steps above failed.
func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string,
func buildPackage(ctx context.Context, fissionClient *crd.FissionClient, envBuilderNamespace string,
storageSvcUrl string, pkg *crd.Package) (uploadResp *fission.ArchiveUploadResponse, buildLogs string, err error) {
env, err := fissionClient.Environments(pkg.Spec.Environment.Namespace).Get(pkg.Spec.Environment.Name)
@@ -60,7 +61,7 @@ func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string,
}
// send fetch request to fetcher
err = fetcherC.Fetch(fetchReq)
err = fetcherC.Fetch(ctx, fetchReq)
if err != nil {
e := fmt.Sprintf("Error fetching source package: %v", err)
log.Println(e)
@@ -103,7 +104,7 @@ func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string,
log.Printf("Start uploading deployment package: %v", buildResp.ArtifactFilename)
// ask fetcher to upload the deployment package
uploadResp, err = fetcherC.Upload(uploadReq)
uploadResp, err = fetcherC.Upload(ctx, uploadReq)
if err != nil {
e := fmt.Sprintf("Error uploading deployment package: %v", err)
log.Println(e)
buildermgr/envwatcher.go
+4
View File
@@ -82,6 +82,7 @@ type (
fetcherImagePullPolicy apiv1.PullPolicy
builderImagePullPolicy apiv1.PullPolicy
useIstio bool
collectorEndpoint string
}
)
@@ -105,6 +106,7 @@ func makeEnvironmentWatcher(fissionClient *crd.FissionClient,
fetcherImagePullPolicy := fission.GetImagePullPolicy(os.Getenv("FETCHER_IMAGE_PULL_POLICY"))
builderImagePullPolicy := fission.GetImagePullPolicy(os.Getenv("BUILDER_IMAGE_PULL_POLICY"))
collectorEndpoint := os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT")
envWatcher := &environmentWatcher{
cache: make(map[string]*builderInfo),
@@ -116,6 +118,7 @@ func makeEnvironmentWatcher(fissionClient *crd.FissionClient,
fetcherImagePullPolicy: fetcherImagePullPolicy,
builderImagePullPolicy: builderImagePullPolicy,
useIstio: useIstio,
collectorEndpoint: collectorEndpoint,
}
go envWatcher.service()
@@ -590,6 +593,7 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment, ns
Command: []string{"/fetcher",
"-secret-dir", sharedSecretPath,
"-cfgmap-dir", sharedCfgMapPath,
"-jaeger-collector-endpoint", envw.collectorEndpoint,
sharedMountPath},
ReadinessProbe: &apiv1.Probe{
InitialDelaySeconds: 5,
buildermgr/pkgwatcher.go
+3 -1
View File
@@ -17,6 +17,7 @@ limitations under the License.
package buildermgr
import (
"context"
"fmt"
"log"
"time"
@@ -160,7 +161,8 @@ func (pkgw *packageWatcher) build(buildCache *cache.Cache, srcpkg *crd.Package)
log.Printf("Setup rolebinding for sa : %s.%s for pkg : %s.%s", fission.FissionBuilderSA, builderNs, pkg.Metadata.Name, pkg.Metadata.Namespace)
}
uploadResp, buildLogs, err := buildPackage(pkgw.fissionClient, builderNs, pkgw.storageSvcUrl, pkg)
ctx := context.Background()
uploadResp, buildLogs, err := buildPackage(ctx, pkgw.fissionClient, builderNs, pkgw.storageSvcUrl, pkg)
if err != nil {
log.Printf("Error building package %v: %v", pkg.Metadata.Name, err)
updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil)
charts/fission-all/templates/deployment.yaml
+33 -10
View File
@@ -131,10 +131,12 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--controllerPort", "8888"]
args: ["--controllerPort", "8888", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: FISSION_FUNCTION_NAMESPACE
value: "{{ .Values.functionNamespace }}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: POD_NAMESPACE
valueFrom:
fieldRef:
@@ -185,7 +187,7 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}"]
args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: POD_NAMESPACE
valueFrom:
@@ -205,6 +207,8 @@ spec:
value: {{ .Values.routerRoundTripSvcAddressUpdateTimeout | default 30 | quote }}
- name: DEBUG_ENV
value: {{ .Values.debugEnv | quote }}
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
readinessProbe:
httpGet:
path: "/router-healthz"
@@ -264,7 +268,7 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"]
args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: FETCHER_IMAGE
value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
@@ -272,6 +276,8 @@ spec:
value: "{{ .Values.pullPolicy }}"
- name: RUNTIME_IMAGE_PULL_POLICY
value: "{{ .Values.pullPolicy }}"
- name: TRACE_JAEGER_COLLECTOR_ENDPOINT
value: "{{ .Values.traceCollectorEndpoint }}"
- name: ENABLE_ISTIO
value: "{{ .Values.enableIstio }}"
- name: FETCHER_MINCPU
@@ -281,7 +287,9 @@ spec:
- name: FETCHER_MAXCPU
value: {{ .Values.fetcherMaxCpu | default "1000m" | quote }}
- name: FETCHER_MAXMEM
value: {{ .Values.fetcherMaxMem | default "128Mi" | quote }}
value: {{ .Values.fetcherMaxMem | default "128Mi" | quote }}
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
readinessProbe:
httpGet:
path: "/healthz"
@@ -321,7 +329,7 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"]
args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: FETCHER_IMAGE
value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
@@ -329,8 +337,12 @@ spec:
value: "{{ .Values.pullPolicy }}"
- name: BUILDER_IMAGE_PULL_POLICY
value: "{{ .Values.pullPolicy }}"
- name: TRACE_JAEGER_COLLECTOR_ENDPOINT
value: "{{ .Values.traceCollectorEndpoint }}"
- name: ENABLE_ISTIO
value: "{{ .Values.enableIstio }}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
---
@@ -352,7 +364,10 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--kubewatcher", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--kubewatcher", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
---
@@ -540,12 +555,14 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: MESSAGE_QUEUE_TYPE
value: nats-streaming
- name: MESSAGE_QUEUE_URL
value: nats://{{ .Values.nats.authToken }}@nats-streaming:4222
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
{{- end }}
@@ -570,7 +587,7 @@ spec:
image: "{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: MESSAGE_QUEUE_TYPE
value: kafka
@@ -578,6 +595,8 @@ spec:
value: "{{.Values.kafka.brokers}}"
- name: MESSAGE_QUEUE_KAFKA_VERSION
value: "{{.Values.kafka.version}}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
{{- end }}
@@ -602,8 +621,10 @@ spec:
image: "{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: MESSAGE_QUEUE_TYPE
value: azure-storage-queue
- name: AZURE_STORAGE_ACCOUNT_NAME
@@ -635,8 +656,10 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--storageServicePort", "8000", "--filePath", "/fission"]
args: ["--storageServicePort", "8000", "--filePath", "/fission", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: PRUNE_INTERVAL
value: "{{.Values.pruneInterval}}"
volumeMounts:
charts/fission-all/values.yaml
+6
View File
@@ -124,9 +124,15 @@ preUpgradeChecksImage: fission/pre-upgrade-checks
## summary is returned as part of http response
debugEnv: true
## set this flag to true if prometheus needs to be deployed along with fission
prometheusDeploy: true
## set this flag to false if you dont need canary deployment feature
canaryDeployment:
enabled: true
# Use these flags to enable opentracing, the variable is endpoint of Jaeger collector in the format shown below
#traceCollectorEndpoint: "http://jaeger-collector.jaeger.svc:14268/api/traces?format=jaeger.thrift"
#traceSamplingRate: 0.75
charts/fission-core/templates/deployment.yaml
+27 -7
View File
@@ -132,8 +132,10 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--controllerPort", "8888"]
args: ["--controllerPort", "8888", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: FISSION_FUNCTION_NAMESPACE
value: "{{ .Values.functionNamespace }}"
- name: POD_NAMESPACE
@@ -183,12 +185,14 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}"]
args: ["--routerPort", "8888", "--executorUrl", "http://executor.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: POD_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: ROUTER_ROUND_TRIP_TIMEOUT
value: {{ .Values.routerRoundTripTimeout | default "50ms" | quote }}
- name: ROUTER_ROUNDTRIP_TIMEOUT_EXPONENT
@@ -253,14 +257,18 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}"]
args: ["--executorPort", "8888", "--namespace", "{{ .Values.functionNamespace }}", "--fission-namespace", "{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: FETCHER_IMAGE
value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
- name: FETCHER_IMAGE_PULL_POLICY
value: "{{ .Values.pullPolicy }}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: ENABLE_ISTIO
value: "{{ .Values.enableIstio }}"
- name: TRACE_JAEGER_COLLECTOR_ENDPOINT
value: "{{ .Values.traceCollectorEndpoint }}"
- name: FETCHER_MINCPU
value: {{ .Values.fetcherMinCpu | default "10m" | quote }}
- name: FETCHER_MINMEM
@@ -303,7 +311,7 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"]
args: ["--builderMgr", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: FETCHER_IMAGE
value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
@@ -311,6 +319,10 @@ spec:
value: "{{ .Values.pullPolicy }}"
- name: BUILDER_IMAGE_PULL_POLICY
value: "{{ .Values.pullPolicy }}"
- name: TRACE_JAEGER_COLLECTOR_ENDPOINT
value: "{{ .Values.traceCollectorEndpoint }}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
- name: ENABLE_ISTIO
value: "{{ .Values.enableIstio }}"
serviceAccount: fission-svc
@@ -334,7 +346,10 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--kubewatcher", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--kubewatcher", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
---
@@ -356,7 +371,10 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--timer", "--routerUrl", "http://router.{{ .Release.Namespace }}"]
args: ["--timer", "--routerUrl", "http://router.{{ .Release.Namespace }}", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
serviceAccount: fission-svc
---
@@ -379,10 +397,12 @@ spec:
image: "{{ .Values.repository }}/{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
command: ["/fission-bundle"]
args: ["--storageServicePort", "8000", "--filePath", "/fission"]
args: ["--storageServicePort", "8000", "--filePath", "/fission", "--collectorEndpoint", "{{ .Values.traceCollectorEndpoint }}"]
env:
- name: PRUNE_INTERVAL
value: "{{.Values.pruneInterval}}"
- name: TRACING_SAMPLING_RATE
value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
volumeMounts:
- name: fission-storage
mountPath: /fission
charts/fission-core/values.yaml
+4
View File
@@ -93,3 +93,7 @@ prometheusDeploy: false
## set this flag to false if you dont need canary deployment feature
canaryDeployment:
enabled: false
# Use these flags to enable opentracing, the variable is endpoint of Jaeger collector in the format shown below
#traceCollectorEndpoint: "http://jaeger-collector.jaeger.svc:14268/api/traces?format=jaeger.thrift"
#traceSamplingRate: 0.75
environments/fetcher/client/client.go
+17 -9
View File
@@ -2,6 +2,7 @@ package client
import (
"bytes"
"context"
"encoding/json"
"io/ioutil"
"log"
@@ -9,18 +10,25 @@ import (
"strings"
"time"
"go.opencensus.io/plugin/ochttp"
"golang.org/x/net/context/ctxhttp"
"github.com/fission/fission"
)
type (
Client struct {
url string
url string
httpClient *http.Client
}
)
func MakeClient(fetcherUrl string) *Client {
return &Client{
url: strings.TrimSuffix(fetcherUrl, "/"),
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
}
}
@@ -36,18 +44,18 @@ func (c *Client) getUploadUrl() string {
return c.url + "/upload"
}
func (c *Client) Specialize(req *fission.FunctionSpecializeRequest) error {
_, err := sendRequest(req, c.getSpecializeUrl())
func (c *Client) Specialize(ctx context.Context, req *fission.FunctionSpecializeRequest) error {
_, err := sendRequest(ctx, c.httpClient, req, c.getSpecializeUrl())
return err
}
func (c *Client) Fetch(fr *fission.FunctionFetchRequest) error {
_, err := sendRequest(fr, c.getFetchUrl())
func (c *Client) Fetch(ctx context.Context, fr *fission.FunctionFetchRequest) error {
_, err := sendRequest(ctx, c.httpClient, fr, c.getFetchUrl())
return err
}
func (c *Client) Upload(fr *fission.ArchiveUploadRequest) (*fission.ArchiveUploadResponse, error) {
body, err := sendRequest(fr, c.getUploadUrl())
func (c *Client) Upload(ctx context.Context, fr *fission.ArchiveUploadRequest) (*fission.ArchiveUploadResponse, error) {
body, err := sendRequest(ctx, c.httpClient, fr, c.getUploadUrl())
uploadResp := fission.ArchiveUploadResponse{}
err = json.Unmarshal(body, &uploadResp)
@@ -58,7 +66,7 @@ func (c *Client) Upload(fr *fission.ArchiveUploadRequest) (*fission.ArchiveUploa
return &uploadResp, nil
}
func sendRequest(req interface{}, url string) ([]byte, error) {
func sendRequest(ctx context.Context, httpClient *http.Client, req interface{}, url string) ([]byte, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, err
@@ -68,7 +76,7 @@ func sendRequest(req interface{}, url string) ([]byte, error) {
var resp *http.Response
for i := 0; i < maxRetries; i++ {
resp, err = http.Post(url, "application/json", bytes.NewReader(body))
resp, err = ctxhttp.Post(ctx, httpClient, url, "application/json", bytes.NewReader(body))
if err == nil {
if resp.StatusCode == 200 {
environments/fetcher/cmd/main.go
+38 -2
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
@@ -11,6 +12,10 @@ import (
"runtime/debug"
"syscall"
"go.opencensus.io/exporter/jaeger"
"go.opencensus.io/plugin/ochttp"
"go.opencensus.io/trace"
"github.com/fission/fission"
"github.com/fission/fission/environments/fetcher"
)
@@ -19,6 +24,29 @@ func dumpStackTrace() {
debug.PrintStack()
}
func registerTraceExporter(collectorEndpoint string) error {
if collectorEndpoint == "" {
return nil
}
serviceName := "Fission-Fetcher"
exporter, err := jaeger.NewExporter(jaeger.Options{
CollectorEndpoint: collectorEndpoint,
Process: jaeger.Process{
ServiceName: serviceName,
Tags: []jaeger.Tag{
jaeger.BoolTag("fission", true),
},
},
})
if err != nil {
return err
}
trace.RegisterExporter(exporter)
trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
return nil
}
// Usage: fetcher <shared volume path>
func main() {
// register signal handler for dumping stack trace.
@@ -32,6 +60,7 @@ func main() {
}()
flag.Usage = fetcherUsage
collectorEndpoint := flag.String("jaeger-collector-endpoint", "", "")
specializeOnStart := flag.Bool("specialize-on-startup", false, "Flag to activate specialize process at pod starup")
specializePayload := flag.String("specialize-request", "", "JSON payload for specialize request")
secretDir := flag.String("secret-dir", "", "Path to shared secrets directory")
@@ -53,6 +82,10 @@ func main() {
}
}
if err := registerTraceExporter(*collectorEndpoint); err != nil {
log.Fatalf("Could not register trace exporter: %v", err)
}
f, err := fetcher.MakeFetcher(dir, *secretDir, *configDir)
if err != nil {
log.Fatalf("Error making fetcher: %v", err)
@@ -70,7 +103,8 @@ func main() {
log.Fatalf("Error decoding specialize request: %v", err)
}
err = f.SpecializePod(specializeReq.FetchReq, specializeReq.LoadReq)
ctx := context.Background()
err = f.SpecializePod(ctx, specializeReq.FetchReq, specializeReq.LoadReq)
if err != nil {
log.Fatalf("Error specialing function poadt: %v", err)
}
@@ -96,7 +130,9 @@ func main() {
})
log.Println("Fetcher ready to receive requests")
http.ListenAndServe(":8000", mux)
http.ListenAndServe(":8000", &ochttp.Handler{
Handler: mux,
})
}
func fetcherUsage() {
environments/fetcher/fetcher.go
+28 -19
View File
@@ -2,6 +2,7 @@ package fetcher
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
@@ -18,7 +19,9 @@ import (
"github.com/mholt/archiver"
"github.com/pkg/errors"
"github.com/satori/go.uuid"
uuid "github.com/satori/go.uuid"
"go.opencensus.io/plugin/ochttp"
"golang.org/x/net/context/ctxhttp"
k8serr "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
@@ -35,6 +38,7 @@ type (
sharedConfigPath string
fissionClient *crd.FissionClient
kubeClient *kubernetes.Clientset
httpClient *http.Client
}
)
@@ -60,11 +64,14 @@ func MakeFetcher(sharedVolumePath string, sharedSecretPath string, sharedConfigP
sharedConfigPath: sharedConfigPath,
fissionClient: fissionClient,
kubeClient: kubeClient,
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
}, nil
}
func downloadUrl(url string, localPath string) error {
resp, err := http.Get(url)
func downloadUrl(ctx context.Context, httpClient *http.Client, url string, localPath string) error {
resp, err := ctxhttp.Get(ctx, httpClient, url)
if err != nil {
return err
}
@@ -116,15 +123,11 @@ func getChecksum(path string) (*fission.Checksum, error) {
}, nil
}
func verifyChecksum(path string, checksum *fission.Checksum) error {
func verifyChecksum(fileChecksum, checksum *fission.Checksum) error {
if checksum.Type != fission.ChecksumTypeSHA256 {
return fission.MakeError(fission.ErrorInvalidArgument, "Unsupported checksum type")
}
c, err := getChecksum(path)
if err != nil {
return err
}
if c.Sum != checksum.Sum {
if fileChecksum.Sum != checksum.Sum {
return fission.MakeError(fission.ErrorChecksumFail, "Checksum validation failed")
}
return nil
@@ -175,8 +178,7 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) {
return
}
log.Printf("fetcher received fetch request and started downloading: %v", req)
code, err := fetcher.Fetch(req)
code, err := fetcher.Fetch(r.Context(), req)
if err != nil {
http.Error(w, err.Error(), code)
return
@@ -217,7 +219,7 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request
//log.Printf("fetcher received fetch request and started downloading: %v", req)
err = fetcher.SpecializePod(req.FetchReq, req.LoadReq)
err = fetcher.SpecializePod(r.Context(), req.FetchReq, req.LoadReq)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
@@ -229,7 +231,7 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request
// Fetch takes FetchRequest and makes the fetch call
// It returns the HTTP code and error if any
func (fetcher *Fetcher) Fetch(req fission.FunctionFetchRequest) (int, error) {
func (fetcher *Fetcher) Fetch(ctx context.Context, req fission.FunctionFetchRequest) (int, error) {
// check that the requested filename is not an empty string and error out if so
if len(req.Filename) == 0 {
e := fmt.Sprintf("Fetch request received for an empty file name, request: %v", req)
@@ -248,7 +250,7 @@ func (fetcher *Fetcher) Fetch(req fission.FunctionFetchRequest) (int, error) {
if req.FetchType == fission.FETCH_URL {
// fetch the file and save it to the tmp path
err := downloadUrl(req.Url, tmpPath)
err := downloadUrl(ctx, fetcher.httpClient, req.Url, tmpPath)
if err != nil {
e := fmt.Sprintf("Failed to download url %v: %v", req.Url, err)
log.Printf(e)
@@ -289,14 +291,21 @@ func (fetcher *Fetcher) Fetch(req fission.FunctionFetchRequest) (int, error) {
}
} else {
// download and verify
err = downloadUrl(archive.URL, tmpPath)
err := downloadUrl(ctx, fetcher.httpClient, archive.URL, tmpPath)
if err != nil {
e := fmt.Sprintf("Failed to download url %v: %v", req.Url, err)
log.Printf(e)
return http.StatusBadRequest, errors.New(e)
}
err = verifyChecksum(tmpPath, &archive.Checksum)
checksum, err := getChecksum(tmpPath)
if err != nil {
e := fmt.Sprintf("Failed to get checksum: %v", err)
log.Printf(e)
return http.StatusBadRequest, errors.New(e)
}
err = verifyChecksum(checksum, &archive.Checksum)
if err != nil {
e := fmt.Sprintf("Failed to verify checksum: %v", err)
log.Printf(e)
@@ -454,7 +463,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) {
log.Println("Starting upload...")
ssClient := storageSvcClient.MakeClient(req.StorageSvcUrl)
fileID, err := ssClient.Upload(dstFilepath, nil)
fileID, err := ssClient.Upload(r.Context(), dstFilepath, nil)
if err != nil {
e := fmt.Sprintf("Error uploading zip file: %v", err)
log.Println(e)
@@ -526,14 +535,14 @@ func (fetcher *Fetcher) unarchive(src string, dst string) error {
return nil
}
func (fetcher *Fetcher) SpecializePod(fetchReq fission.FunctionFetchRequest, loadReq fission.FunctionLoadRequest) error {
func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq fission.FunctionFetchRequest, loadReq fission.FunctionLoadRequest) error {
startTime := time.Now()
defer func() {
elapsed := time.Since(startTime)
log.Printf("Elapsed time in fetch request = %v", elapsed)
}()
_, err := fetcher.Fetch(fetchReq)
_, err := fetcher.Fetch(ctx, fetchReq)
if err != nil {
return errors.Wrap(err, "Error fetching deploy package")
}
executor/api.go
+9 -3
View File
@@ -26,6 +26,7 @@ import (
"strings"
"github.com/gorilla/mux"
"go.opencensus.io/plugin/ochttp"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
@@ -46,7 +47,7 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
return
}
serviceName, err := executor.getServiceForFunction(&m)
serviceName, err := executor.getServiceForFunction(r.Context(), &m)
if err != nil {
code, msg := fission.GetHTTPError(err)
log.Printf("Error: %v: %v", code, msg)
@@ -66,7 +67,7 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
// stale addresses are not returned to the router.
// To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and
// invalidates the cache entry if the pod address was cached.
func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, error) {
func (executor *Executor) getServiceForFunction(ctx context.Context, m *metav1.ObjectMeta) (string, error) {
// Check function -> svc cache
log.Printf("[%v] Checking for cached function service", m.Name)
fsvc, err := executor.fsCache.GetByFunction(m)
@@ -82,6 +83,7 @@ func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, e
respChan := make(chan *createFuncServiceResponse)
executor.requestChan <- &createFuncServiceRequest{
ctx: ctx,
funcMeta: m,
respChan: respChan,
}
@@ -127,5 +129,9 @@ func (executor *Executor) Serve(port int) {
executor.ndm.Run(ctx)
executor.gpm.Run(ctx)
r.Use(fission.LoggingMiddleware)
log.Fatal(http.ListenAndServe(address, r))
err := http.ListenAndServe(address, &ochttp.Handler{
Handler: r,
// Propagation: &b3.HTTPFormat{},
})
log.Fatal(err)
}
executor/client/client.go
+9 -2
View File
@@ -18,6 +18,7 @@ package client
import (
"bytes"
"context"
"encoding/json"
"io/ioutil"
"log"
@@ -26,6 +27,8 @@ import (
"strings"
"time"
"go.opencensus.io/plugin/ochttp"
"golang.org/x/net/context/ctxhttp"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
@@ -35,6 +38,7 @@ type Client struct {
executorUrl string
tappedByUrl map[string]bool
requestChan chan string
httpClient *http.Client
}
func MakeClient(executorUrl string) *Client {
@@ -42,12 +46,15 @@ func MakeClient(executorUrl string) *Client {
executorUrl: strings.TrimSuffix(executorUrl, "/"),
tappedByUrl: make(map[string]bool),
requestChan: make(chan string),
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
}
go c.service()
return c
}
func (c *Client) GetServiceForFunction(metadata *metav1.ObjectMeta) (string, error) {
func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.ObjectMeta) (string, error) {
executorUrl := c.executorUrl + "/v2/getServiceForFunction"
body, err := json.Marshal(metadata)
@@ -55,7 +62,7 @@ func (c *Client) GetServiceForFunction(metadata *metav1.ObjectMeta) (string, err
return "", err
}
resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body))
resp, err := ctxhttp.Post(ctx, c.httpClient, executorUrl, "application/json", bytes.NewReader(body))
if err != nil {
return "", err
}
executor/executor.go
+6 -4
View File
@@ -17,6 +17,7 @@ limitations under the License.
package executor
import (
"context"
"fmt"
"log"
"net/http"
@@ -50,6 +51,7 @@ type (
fsCreateWg map[string]*sync.WaitGroup
}
createFuncServiceRequest struct {
ctx context.Context
funcMeta *metav1.ObjectMeta
respChan chan *createFuncServiceResponse
}
@@ -98,7 +100,7 @@ func (executor *Executor) serveCreateFuncServices() {
// launch a goroutine for each request, to parallelize
// the specialization of different functions
go func() {
fsvc, err := executor.createServiceForFunction(m)
fsvc, err := executor.createServiceForFunction(req.ctx, m)
req.respChan <- &createFuncServiceResponse{
funcSvc: fsvc,
err: err,
@@ -137,7 +139,7 @@ func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fiss
return fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, nil
}
func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
func (executor *Executor) createServiceForFunction(ctx context.Context, meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
log.Printf("[%v] No cached function service found, creating one", meta.Name)
executorType, err := executor.getFunctionExecutorType(meta)
@@ -150,9 +152,9 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs
switch executorType {
case fission.ExecutorTypeNewdeploy:
fsvc, fsvcErr = executor.ndm.GetFuncSvc(meta)
fsvc, fsvcErr = executor.ndm.GetFuncSvc(ctx, meta)
default:
fsvc, fsvcErr = executor.gpm.GetFuncSvc(meta)
fsvc, fsvcErr = executor.gpm.GetFuncSvc(ctx, meta)
}
if fsvcErr != nil {
executor/executor_test.go
+2 -1
View File
@@ -23,6 +23,7 @@
package executor
import (
"context"
"fmt"
"io/ioutil"
"log"
@@ -237,7 +238,7 @@ func TestExecutor(t *testing.T) {
// the main test: get a service for a given function
t1 := time.Now()
svc, err := poolmgrClient.GetServiceForFunction(&f.Metadata)
svc, err := poolmgrClient.GetServiceForFunction(context.Background(), &f.Metadata)
if err != nil {
log.Panicf("failed to get func svc: %v", err)
}
executor/newdeploy/newdeploy.go
+1
View File
@@ -292,6 +292,7 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen
"-specialize-request", string(specializePayload),
"-secret-dir", deploy.sharedSecretPath,
"-cfgmap-dir", deploy.sharedCfgMapPath,
"-jaeger-collector-endpoint", deploy.collectorEndpoint,
deploy.sharedMountPath},
Lifecycle: &apiv1.Lifecycle{
PreStop: &apiv1.Handler{
executor/newdeploy/newdeploymgr.go
+5 -1
View File
@@ -59,6 +59,7 @@ type (
sharedSecretPath string
sharedCfgMapPath string
useIstio bool
collectorEndpoint string
fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and pod name
@@ -85,6 +86,8 @@ func MakeNewDeploy(
fetcherImg = "fission/fetcher"
}
collectorEndpoint := os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT")
enableIstio := false
if len(os.Getenv("ENABLE_ISTIO")) > 0 {
istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO"))
@@ -110,6 +113,7 @@ func MakeNewDeploy(
sharedMountPath: "/userfunc",
sharedSecretPath: "/secrets",
sharedCfgMapPath: "/configs",
collectorEndpoint: collectorEndpoint,
useIstio: enableIstio,
idlePodReapTime: 2 * time.Minute,
@@ -160,7 +164,7 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll
return store, controller
}
func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name)
if err != nil {
return nil, err
executor/poolmgr/gp.go
+10 -5
View File
@@ -17,6 +17,7 @@ limitations under the License.
package poolmgr
import (
"context"
"fmt"
"log"
"math/rand"
@@ -67,6 +68,7 @@ type (
sharedMountPath string // used by generic pool when creating env deployment to specify the share volume path for fetcher & env
sharedSecretPath string
sharedCfgMapPath string
collectorEndpoint string
}
// serialize the choosing of pods so that choices don't conflict
@@ -89,7 +91,8 @@ func MakeGenericPool(
functionNamespace string,
fsCache *fscache.FunctionServiceCache,
instanceId string,
enableIstio bool) (*GenericPool, error) {
enableIstio bool,
collectorEndpoint string) (*GenericPool, error) {
log.Printf("Creating pool for environment %v", env.Metadata)
@@ -119,6 +122,7 @@ func MakeGenericPool(
sharedMountPath: "/userfunc", // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path
sharedSecretPath: "/secrets",
sharedCfgMapPath: "/configs",
collectorEndpoint: collectorEndpoint,
}
gp.runtimeImagePullPolicy = fission.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY"))
@@ -301,7 +305,7 @@ func (gp *GenericPool) getSpecializeUrl(podIP string) string {
// specializePod chooses a pod, copies the required user-defined function to that pod
// (via fetcher), and calls the function-run container to load it, resulting in a
// specialized pod.
func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta) error {
func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, metadata *metav1.ObjectMeta) error {
// for fetcher we don't need to create a service, just talk to the pod directly
podIP := pod.Status.PodIP
if len(podIP) == 0 {
@@ -354,7 +358,7 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta
log.Printf("[%v] specializing pod", metadata.Name)
err = fetcherClient.MakeClient(fetcherUrl).Specialize(&specializeReq)
err = fetcherClient.MakeClient(fetcherUrl).Specialize(ctx, &specializeReq)
if err != nil {
return err
}
@@ -487,6 +491,7 @@ func (gp *GenericPool) createPool() error {
Command: []string{"/fetcher",
"-secret-dir", gp.sharedSecretPath,
"-cfgmap-dir", gp.sharedCfgMapPath,
"-jaeger-collector-endpoint", gp.collectorEndpoint,
gp.sharedMountPath},
// Pod is removed from endpoints list for service when it's
// state became "Termination". We used preStop hook as the
@@ -617,7 +622,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.
return svc, err
}
func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
log.Printf("[%v] Choosing pod from pool", m.Name)
newLabels := gp.labelsForFunction(m)
@@ -667,7 +672,7 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error
return nil, err
}
err = gp.specializePod(pod, m)
err = gp.specializePod(ctx, pod, m)
if err != nil {
gp.scheduleDeletePod(pod.ObjectMeta.Name)
return nil, err
executor/poolmgr/gpm.go
+17 -13
View File
@@ -55,7 +55,9 @@ type (
instanceId string
requestChannel chan *request
enableIstio bool
enableIstio bool
collectorEndpoint string
funcStore k8sCache.Store
funcController k8sCache.Controller
pkgStore k8sCache.Store
@@ -82,15 +84,16 @@ func MakeGenericPoolManager(
instanceId string) *GenericPoolManager {
gpm := &GenericPoolManager{
pools: make(map[string]*GenericPool),
kubernetesClient: kubernetesClient,
namespace: functionNamespace,
fissionClient: fissionClient,
functionEnv: cache.MakeCache(10*time.Second, 0),
fsCache: fscache.MakeFunctionServiceCache(),
instanceId: instanceId,
requestChannel: make(chan *request),
idlePodReapTime: 2 * time.Minute,
pools: make(map[string]*GenericPool),
kubernetesClient: kubernetesClient,
namespace: functionNamespace,
fissionClient: fissionClient,
functionEnv: cache.MakeCache(10*time.Second, 0),
fsCache: fscache.MakeFunctionServiceCache(),
instanceId: instanceId,
requestChannel: make(chan *request),
idlePodReapTime: 2 * time.Minute,
collectorEndpoint: os.Getenv("TRACE_JAEGER_COLLECTOR_ENDPOINT"),
}
go gpm.service()
go gpm.eagerPoolCreator()
@@ -141,7 +144,8 @@ func (gpm *GenericPoolManager) service() {
pool, err = MakeGenericPool(
gpm.fissionClient, gpm.kubernetesClient, req.env, poolsize,
ns, gpm.namespace, gpm.fsCache, gpm.instanceId, gpm.enableIstio)
ns, gpm.namespace, gpm.fsCache, gpm.instanceId, gpm.enableIstio,
gpm.collectorEndpoint)
if err != nil {
req.responseChannel <- &response{error: err}
continue
@@ -189,7 +193,7 @@ func (gpm *GenericPoolManager) CleanupPools(envs []crd.Environment) {
}
}
func (gpm *GenericPoolManager) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
// from Func -> get Env
log.Printf("[%v] getting environment for function", metadata.Name)
env, err := gpm.getFunctionEnv(metadata)
@@ -204,7 +208,7 @@ func (gpm *GenericPoolManager) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache
// from GenericPool -> get one function container
// (this also adds to the cache)
log.Printf("[%v] getting function service from pool", metadata.Name)
return pool.GetFuncSvc(metadata)
return pool.GetFuncSvc(ctx, metadata)
}
func (gpm *GenericPoolManager) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) {
fission-bundle/main.go
+67 -9
View File
@@ -6,7 +6,10 @@ import (
"os"
"strconv"
"github.com/docopt/docopt-go"
"go.opencensus.io/exporter/jaeger"
"go.opencensus.io/trace"
docopt "github.com/docopt/docopt-go"
"github.com/fission/fission"
"github.com/fission/fission/buildermgr"
@@ -91,6 +94,55 @@ func getStringArgWithDefault(arg interface{}, defaultValue string) string {
}
}
func registerTraceExporter(arguments map[string]interface{}) error {
collectorEndpoint := getStringArgWithDefault(arguments["--collectorEndpoint"], "")
if collectorEndpoint == "" {
log.Print("Skipping trace exporter registration")
return nil
}
serviceName := "Fission-Unknown"
if arguments["--controllerPort"] != nil {
serviceName = "Fission-Controller"
} else if arguments["--routerPort"] != nil {
serviceName = "Fission-Router"
} else if arguments["--executorPort"] != nil {
serviceName = "Fission-Executor"
} else if arguments["--kubewatcher"] == true {
serviceName = "Fission-KubeWatcher"
} else if arguments["--timer"] == true {
serviceName = "Fission-Timer"
} else if arguments["--mqt"] == true {
serviceName = "Fission-MessageQueueTrigger"
} else if arguments["--builderMgr"] == true {
serviceName = "Fission-BuilderMgr"
} else if arguments["--storageServicePort"] != nil {
serviceName = "Fission-StorageSvc"
}
exporter, err := jaeger.NewExporter(jaeger.Options{
CollectorEndpoint: collectorEndpoint,
Process: jaeger.Process{
ServiceName: serviceName,
Tags: []jaeger.Tag{
jaeger.BoolTag("fission", true),
},
},
})
if err != nil {
return err
}
samplingRate, err := strconv.ParseFloat(os.Getenv("TRACING_SAMPLING_RATE"), 32)
if err != nil {
return err
}
trace.RegisterExporter(exporter)
trace.ApplyConfig(trace.Config{DefaultSampler: trace.ProbabilitySampler(samplingRate)})
return nil
}
func main() {
usage := `fission-bundle: Package of all fission microservices: controller, router, executor.
@@ -114,16 +166,17 @@ Use it to start one or more of the fission servers:
backends.
Usage:
fission-bundle --controllerPort=<port>
fission-bundle --routerPort=<port> [--executorUrl=<url>]
fission-bundle --executorPort=<port> [--namespace=<namespace>] [--fission-namespace=<namespace>]
fission-bundle --kubewatcher [--routerUrl=<url>]
fission-bundle --storageServicePort=<port> --filePath=<filePath>
fission-bundle --builderMgr [--storageSvcUrl=<url>] [--envbuilder-namespace=<namespace>]
fission-bundle --timer [--routerUrl=<url>]
fission-bundle --mqt [--routerUrl=<url>]
fission-bundle --controllerPort=<port> [--collectorEndpoint=<url>]
fission-bundle --routerPort=<port> [--executorUrl=<url>] [--collectorEndpoint=<url>]
fission-bundle --executorPort=<port> [--namespace=<namespace>] [--fission-namespace=<namespace>] [--collectorEndpoint=<url>]
fission-bundle --kubewatcher [--routerUrl=<url>] [--collectorEndpoint=<url>]
fission-bundle --storageServicePort=<port> --filePath=<filePath> [--collectorEndpoint=<url>]
fission-bundle --builderMgr [--storageSvcUrl=<url>] [--envbuilder-namespace=<namespace>] [--collectorEndpoint=<url>]
fission-bundle --timer [--routerUrl=<url>] [--collectorEndpoint=<url>]
fission-bundle --mqt [--routerUrl=<url>] [--collectorEndpoint=<url>]
fission-bundle --version
Options:
--collectorEndpoint=<url> Jaeger HTTP Thrift collector URL.
--controllerPort=<port> Port that the controller should listen on.
--routerPort=<port> Port that the router should listen on.
--executorPort=<port> Port that the executor should listen on.
@@ -146,6 +199,11 @@ Options:
log.Fatalf("Error: %v", err)
}
err = registerTraceExporter(arguments)
if err != nil {
log.Fatalf("Error: %v", err)
}
functionNs := getStringArgWithDefault(arguments["--namespace"], "fission-function")
fissionNs := getStringArgWithDefault(arguments["--fission-namespace"], "fission")
envBuilderNs := getStringArgWithDefault(arguments["--envbuilder-namespace"], "fission-builder")
fission/package.go
+5 -3
View File
@@ -18,6 +18,7 @@ package main
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
@@ -533,10 +534,11 @@ func createArchive(client *client.Client, includeFiles []string, noZip bool, spe
archivePath := makeArchiveFileIfNeeded("", includeFiles, noZip)
return uploadArchive(client, archivePath)
ctx := context.Background()
return uploadArchive(ctx, client, archivePath)
}
func uploadArchive(client *client.Client, fileName string) *fission.Archive {
func uploadArchive(ctx context.Context, client *client.Client, fileName string) *fission.Archive {
var archive fission.Archive
// If filename is a URL, download it first
@@ -552,7 +554,7 @@ func uploadArchive(client *client.Client, fileName string) *fission.Archive {
ssClient := storageSvcClient.MakeClient(u)
// TODO add a progress bar
id, err := ssClient.Upload(fileName, nil)
id, err := ssClient.Upload(ctx, fileName, nil)
util.CheckErr(err, fmt.Sprintf("upload file %v", fileName))
storageSvc, err := client.GetSvcURL("application=fission-storage")
fission/spec.go
+2 -1
View File
@@ -858,7 +858,8 @@ func applyArchives(fclient *client.Client, specDir string, fr *FissionResources)
// doesn't exist, upload
fmt.Printf("uploading archive %v\n", name)
// ar.URL is actually a local filename at this stage
uploadedAr := uploadArchive(fclient, ar.URL)
ctx := context.Background()
uploadedAr := uploadArchive(ctx, fclient, ar.URL)
archiveFiles[name] = *uploadedAr
}
}
fission/upgrade.go
+3 -1
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"encoding/base64"
"encoding/json"
"fmt"
@@ -293,7 +294,8 @@ func upgradeRestoreState(c *cli.Context) error {
tmpfile.Close()
// upload
archive := uploadArchive(client, tmpfile.Name())
ctx := context.Background()
archive := uploadArchive(ctx, client, tmpfile.Name())
os.Remove(tmpfile.Name())
// create pkg
glide.lock
Generated
+37 -4
View File
@@ -1,11 +1,15 @@
hash: 07bc3f35b3b63ff6f15edf3b8724eb61bd6427572322fc171efbe18cec91cd9e
updated: 2018-09-17T14:46:13.995469387+05:30
hash: fbb8e5b2fc64114cf4cd7158a1498f36c580e01b327c14cd35682f999ce13db3
updated: 2019-01-24T19:14:41.16237463+05:30
imports:
- name: cloud.google.com/go
version: 3b1ae45394a234c385be014e9a488f2bb6eef821
subpackages:
- compute/metadata
- internal
- name: git.apache.org/thrift.git
version: 22749ac376b10982eb5fa5a32ba336b20e1e6344
subpackages:
- lib/go/thrift
- name: github.com/Azure/azure-sdk-for-go
version: f111fc2fa3861c5fdced76cae4c9c71821969577
subpackages:
@@ -27,6 +31,8 @@ imports:
version: f87b566248bb0713a56dc55bc545aa5aad17ace0
subpackages:
- client
- name: github.com/DataDog/zstd
version: 1e382f59b41eebd6f592c5db4fd1958ec38a0eba
- name: github.com/davecgh/go-spew
version: 8991bc29aa16c548c550c7ff78260e27b9ab7c73
subpackages:
@@ -156,7 +162,7 @@ imports:
subpackages:
- pb
- name: github.com/nats-io/nats-streaming-server
version: 8910c0c347bc51cc87227aeb27bef19d409bf5c2
version: 7b758bb93407505bc1d0cf26d091fc281d99ca0c
subpackages:
- spb
- util
@@ -173,6 +179,7 @@ imports:
- name: github.com/prometheus/client_golang
version: c5b7fccd204277076155f10851dad72b76a49317
subpackages:
- api/prometheus
- prometheus
- prometheus/promhttp
- name: github.com/prometheus/client_model
@@ -196,7 +203,7 @@ imports:
- name: github.com/satori/go.uuid
version: f58768cc1a7a7e77a3bd49e98cdd21419399b6a3
- name: github.com/Shopify/sarama
version: a6144ae922fd99dd0ea5046c8137acfb7fab0914
version: 03a43f93cd29dc549e6d9b11892795c206f9c38c
- name: github.com/sirupsen/logrus
version: 68cec9f21fbf3ea8d8f98c044bc6ce05f17b267a
- name: github.com/spf13/pflag
@@ -215,6 +222,24 @@ imports:
- lzma
- name: github.com/urfave/cli
version: cfb38830724cc34fedffe9a2a29fb54fa9169cd1
- name: go.opencensus.io
version: 2b5032d79456124f42db6b7eb19ac6c155449dc2
subpackages:
- exemplar
- exporter/jaeger
- exporter/jaeger/internal/gen-go/jaeger
- internal
- internal/tagencoding
- plugin/ochttp
- plugin/ochttp/propagation/b3
- stats
- stats/internal
- stats/view
- tag
- trace
- trace/internal
- trace/propagation
- trace/tracestate
- name: golang.org/x/crypto
version: 81e90905daefcd6fd217b62423c0908922eadb30
subpackages:
@@ -235,6 +260,10 @@ imports:
- internal
- jws
- jwt
- name: golang.org/x/sync
version: 1d60e4601c6fd243af51cc01ddf169918a5407ca
subpackages:
- semaphore
- name: golang.org/x/sys
version: 95c6576299259db960f6c5b9b69ea52422860fce
subpackages:
@@ -255,6 +284,10 @@ imports:
version: 87c7dcbd5db6be1a938380ce9944ed2299806701
subpackages:
- imports
- name: google.golang.org/api
version: 8001663557ac8d144131ea92d9829a0d8e668ab7
subpackages:
- support/bundler
- name: google.golang.org/appengine
version: 9d8544a6b2c7df9cff240fcf92d7b2f59bc13416
repo: https://github.com/golang/appengine
glide.yaml
+3
View File
@@ -74,3 +74,6 @@ import:
version: ^2.1.11
- package: github.com/Shopify/sarama
version: ^1.15.0
- package: go.opencensus.io
version: ^0.19.0
- package: google.golang.org/api/support/bundler
router/functionHandler.go
+24 -6
View File
@@ -18,6 +18,7 @@ package router
import (
"bytes"
"context"
"fmt"
"io"
"io/ioutil"
@@ -33,6 +34,7 @@ import (
"github.com/pkg/errors"
"github.com/satori/go.uuid"
log "github.com/sirupsen/logrus"
"go.opencensus.io/plugin/ochttp"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
@@ -85,6 +87,7 @@ type (
// A layer on top of http.DefaultTransport, with retries.
RetryingRoundTripper struct {
funcHandler *functionHandler
base http.RoundTripper
}
// To keep the request body open during retries, we create an interface with Close operation being a no-op.
@@ -213,7 +216,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
for i := 0; i < roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1; i++ {
// get function service url from cache or executor
serviceUrl, serviceUrlFromCache, err := roundTripper.funcHandler.getServiceEntry()
serviceUrl, serviceUrlFromCache, err := roundTripper.funcHandler.getServiceEntry(req.Context())
if err != nil {
// We might want a specific error code or header for fission failures as opposed to
// user function bugs.
@@ -271,7 +274,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
overhead := time.Since(startTime)
// forward the request to the function service
resp, err = transport.RoundTrip(req)
resp, err = roundTripper.base.RoundTrip(req)
if err == nil {
// Track metrics
httpMetricLabels.code = resp.StatusCode
@@ -411,6 +414,21 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h
Director: director,
Transport: &RetryingRoundTripper{
funcHandler: &fh,
base: &ochttp.Transport{
Base: &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: fh.tsRoundTripperParams.timeout,
KeepAlive: fh.tsRoundTripperParams.keepAlive,
}).DialContext,
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
// Disables caching, Please refer to issue and specifically comment: https://github.com/fission/fission/issues/723#issuecomment-398781995
DisableKeepAlives: true,
},
},
},
}
@@ -494,7 +512,7 @@ func addForwardedHostHeader(req *http.Request) {
}
// getServiceEntry is a short-hand for developers to get service url entry that may returns from executor or cache
func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFromCache bool, err error) {
func (fh *functionHandler) getServiceEntry(ctx context.Context) (serviceUrl *url.URL, serviceUrlFromCache bool, err error) {
// try to find service url from cache first
serviceUrl, err = fh.getServiceEntryFromCache()
if err == nil && serviceUrl != nil {
@@ -514,7 +532,7 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro
// Get service entry from executor and update cache if its the first goroutine
if firstToTheLock { // first to the service url
log.Printf("Calling getServiceForFunction for function: %s", fh.function.Name)
u, err = fh.getServiceEntryFromExecutor()
u, err = fh.getServiceEntryFromExecutor(ctx)
if err != nil {
log.Printf("Error getting service url from executor: %v", err)
return nil, err
@@ -573,9 +591,9 @@ func (fh *functionHandler) getServiceEntryFromCache() (serviceUrl *url.URL, err
}
// getServiceEntryFromExecutor returns service url entry returns from executor
func (fh *functionHandler) getServiceEntryFromExecutor() (*url.URL, error) {
func (fh *functionHandler) getServiceEntryFromExecutor(ctx context.Context) (*url.URL, error) {
// send a request to executor to specialize a new pod
service, err := fh.executor.GetServiceForFunction(fh.function)
service, err := fh.executor.GetServiceForFunction(ctx, fh.function)
if err != nil {
statusCode, errMsg := fission.GetHTTPError(err)
log.Printf("Error from GetServiceForFunction for function (%v): %v : %v", fh.function, statusCode, errMsg)
router/router.go
+8 -1
View File
@@ -50,6 +50,8 @@ import (
"github.com/gorilla/mux"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.opencensus.io/plugin/ochttp"
"go.opencensus.io/trace"
"github.com/fission/fission"
"github.com/fission/fission/crd"
@@ -72,7 +74,12 @@ func router(ctx context.Context, httpTriggerSet *HTTPTriggerSet, resolver *funct
func serve(ctx context.Context, port int, httpTriggerSet *HTTPTriggerSet, resolver *functionReferenceResolver) {
mr := router(ctx, httpTriggerSet, resolver)
url := fmt.Sprintf(":%v", port)
http.ListenAndServe(url, mr)
http.ListenAndServe(url, &ochttp.Handler{
Handler: mr,
StartOptions: trace.StartOptions{
Sampler: trace.AlwaysSample(),
},
})
}
func serveMetric() {
storagesvc/client/client.go
+15 -11
View File
@@ -18,6 +18,7 @@ package client
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
@@ -29,12 +30,16 @@ import (
"os"
"strings"
"go.opencensus.io/plugin/ochttp"
"golang.org/x/net/context/ctxhttp"
"github.com/fission/fission/storagesvc"
)
type (
Client struct {
url string
url string
httpClient *http.Client
}
)
@@ -42,13 +47,16 @@ type (
func MakeClient(url string) *Client {
return &Client{
url: strings.TrimSuffix(url, "/") + "/v1",
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
}
}
// Upload sends the local file pointed to by filePath to the storage
// service, along with the metadata. It returns a file ID that can be
// used to retrieve the file.
func (c *Client) Upload(filePath string, metadata *map[string]string) (string, error) {
func (c *Client) Upload(ctx context.Context, filePath string, metadata *map[string]string) (string, error) {
fi, err := os.Stat(filePath)
if err != nil {
return "", err
@@ -82,9 +90,7 @@ func (c *Client) Upload(filePath string, metadata *map[string]string) (string, e
req.Header["X-File-Size"] = []string{fmt.Sprintf("%v", fileSize)}
req.Header["Content-Type"] = []string{contentType}
client := &http.Client{}
resp, err := client.Do(req)
resp, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return "", err
}
@@ -114,7 +120,7 @@ func (c *Client) GetUrl(id string) string {
// Download fetches the file identified by ID to the local file path.
// filePath must not exist.
func (c *Client) Download(id string, filePath string) error {
func (c *Client) Download(ctx context.Context, id string, filePath string) error {
// url for id
url := c.GetUrl(id)
@@ -132,7 +138,7 @@ func (c *Client) Download(id string, filePath string) error {
defer f.Close()
// make request
resp, err := http.Get(url)
resp, err := ctxhttp.Get(ctx, c.httpClient, url)
if err != nil {
fmt.Println(err)
os.Remove(filePath)
@@ -154,7 +160,7 @@ func (c *Client) Download(id string, filePath string) error {
return nil
}
func (c *Client) Delete(id string) error {
func (c *Client) Delete(ctx context.Context, id string) error {
url := c.GetUrl(id)
req, err := http.NewRequest(http.MethodDelete, url, nil)
@@ -162,9 +168,7 @@ func (c *Client) Delete(id string) error {
return err
}
client := &http.Client{}
resp, err := client.Do(req)
resp, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return err
}
storagesvc/client/storagesvc_test.go
+6 -4
View File
@@ -18,6 +18,7 @@ package client
import (
"bytes"
"context"
"fmt"
"io/ioutil"
"log"
@@ -64,7 +65,8 @@ func TestStorageService(t *testing.T) {
// store it
metadata := make(map[string]string)
fileId, err := client.Upload(tmpfile.Name(), &metadata)
ctx := context.Background()
fileId, err := client.Upload(ctx, tmpfile.Name(), &metadata)
panicIf(err)
// make a temp file for verification
@@ -73,7 +75,7 @@ func TestStorageService(t *testing.T) {
os.Remove(retrievedfile.Name())
// retrieve uploaded file
err = client.Download(fileId, retrievedfile.Name())
err = client.Download(ctx, fileId, retrievedfile.Name())
panicIf(err)
defer os.Remove(retrievedfile.Name())
@@ -87,11 +89,11 @@ func TestStorageService(t *testing.T) {
}
// delete uploaded file
err = client.Delete(fileId)
err = client.Delete(ctx, fileId)
panicIf(err)
// make sure download fails
err = client.Download(fileId, "xxx")
err = client.Download(ctx, fileId, "xxx")
if err == nil {
log.Panic("Download succeeded but file isn't supposed to exist")
}
storagesvc/storagesvc.go
+7 -1
View File
@@ -29,6 +29,7 @@ import (
"github.com/gorilla/mux"
_ "github.com/graymeta/stow/local"
log "github.com/sirupsen/logrus"
"go.opencensus.io/plugin/ochttp"
)
type (
@@ -170,7 +171,12 @@ func (ss *StorageService) Start(port int) {
address := fmt.Sprintf(":%v", port)
r.Use(fission.LoggingMiddleware)
log.Fatal(http.ListenAndServe(address, r))
err := http.ListenAndServe(address, &ochttp.Handler{
Handler: r,
// Propagation: &b3.HTTPFormat{},
})
log.Fatal(err)
}
func RunStorageService(storageType StorageType, storagePath string, containerName string, port int, enablePruner bool) *StorageService {