From 5d2abdd95b872873ee4ecbb3706cdd45af8805dd Mon Sep 17 00:00:00 2001 From: Vishal Date: Thu, 7 Feb 2019 11:56:41 +0530 Subject: [PATCH] OpenTracing for Fission (#1079) Added Opentracing integration using opencensus libraries for all Fission components. --- buildermgr/common.go | 7 +- buildermgr/envwatcher.go | 4 + buildermgr/pkgwatcher.go | 4 +- charts/fission-all/templates/deployment.yaml | 43 ++++++++--- charts/fission-all/values.yaml | 6 ++ charts/fission-core/templates/deployment.yaml | 34 +++++++-- charts/fission-core/values.yaml | 4 + environments/fetcher/client/client.go | 26 ++++--- environments/fetcher/cmd/main.go | 40 +++++++++- environments/fetcher/fetcher.go | 47 +++++++----- executor/api.go | 12 ++- executor/client/client.go | 11 ++- executor/executor.go | 10 ++- executor/executor_test.go | 3 +- executor/newdeploy/newdeploy.go | 1 + executor/newdeploy/newdeploymgr.go | 6 +- executor/poolmgr/gp.go | 15 ++-- executor/poolmgr/gpm.go | 30 ++++---- fission-bundle/main.go | 76 ++++++++++++++++--- fission/package.go | 8 +- fission/spec.go | 3 +- fission/upgrade.go | 4 +- glide.lock | 41 +++++++++- glide.yaml | 3 + router/functionHandler.go | 30 ++++++-- router/router.go | 9 ++- storagesvc/client/client.go | 26 ++++--- storagesvc/client/storagesvc_test.go | 10 ++- storagesvc/storagesvc.go | 8 +- 29 files changed, 400 insertions(+), 121 deletions(-) diff --git a/buildermgr/common.go b/buildermgr/common.go index 833242d5..2f24182c 100644 --- a/buildermgr/common.go +++ b/buildermgr/common.go @@ -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) diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go index b5a6f449..571547c7 100644 --- a/buildermgr/envwatcher.go +++ b/buildermgr/envwatcher.go @@ -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, diff --git a/buildermgr/pkgwatcher.go b/buildermgr/pkgwatcher.go index 7cd1c461..21b1444d 100644 --- a/buildermgr/pkgwatcher.go +++ b/buildermgr/pkgwatcher.go @@ -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) diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 4fd8ae2f..2ec08274 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -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: diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index 60b52504..9a27c86b 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -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 \ No newline at end of file diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 1517ea1c..c3daa386 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -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 diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index 7b36dad0..1e5a94da 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -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 \ No newline at end of file diff --git a/environments/fetcher/client/client.go b/environments/fetcher/client/client.go index 7b68a707..81ff5985 100644 --- a/environments/fetcher/client/client.go +++ b/environments/fetcher/client/client.go @@ -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 { diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 2e6498f7..cfdbc921 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -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 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() { diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index 8d3d13d3..7ef694d5 100644 --- a/environments/fetcher/fetcher.go +++ b/environments/fetcher/fetcher.go @@ -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") } diff --git a/executor/api.go b/executor/api.go index 80121e3b..3bf51223 100644 --- a/executor/api.go +++ b/executor/api.go @@ -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) } diff --git a/executor/client/client.go b/executor/client/client.go index 1b2802d8..803cb958 100644 --- a/executor/client/client.go +++ b/executor/client/client.go @@ -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 } diff --git a/executor/executor.go b/executor/executor.go index 7520900b..6ad7c95d 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -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 { diff --git a/executor/executor_test.go b/executor/executor_test.go index f7bb1b3c..c99ce473 100644 --- a/executor/executor_test.go +++ b/executor/executor_test.go @@ -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) } diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 34c174d3..18575524 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -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{ diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index bace001e..53187a76 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -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 diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index 6563adec..f3965d39 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -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 diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index e197398c..a38bb8fd 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -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) { diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 8b189c91..2d27d938 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -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= - fission-bundle --routerPort= [--executorUrl=] - fission-bundle --executorPort= [--namespace=] [--fission-namespace=] - fission-bundle --kubewatcher [--routerUrl=] - fission-bundle --storageServicePort= --filePath= - fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] - fission-bundle --timer [--routerUrl=] - fission-bundle --mqt [--routerUrl=] + fission-bundle --controllerPort= [--collectorEndpoint=] + fission-bundle --routerPort= [--executorUrl=] [--collectorEndpoint=] + fission-bundle --executorPort= [--namespace=] [--fission-namespace=] [--collectorEndpoint=] + fission-bundle --kubewatcher [--routerUrl=] [--collectorEndpoint=] + fission-bundle --storageServicePort= --filePath= [--collectorEndpoint=] + fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] [--collectorEndpoint=] + fission-bundle --timer [--routerUrl=] [--collectorEndpoint=] + fission-bundle --mqt [--routerUrl=] [--collectorEndpoint=] fission-bundle --version Options: + --collectorEndpoint= Jaeger HTTP Thrift collector URL. --controllerPort= Port that the controller should listen on. --routerPort= Port that the router should listen on. --executorPort= 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") diff --git a/fission/package.go b/fission/package.go index 88fa2d29..c649ff0b 100644 --- a/fission/package.go +++ b/fission/package.go @@ -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") diff --git a/fission/spec.go b/fission/spec.go index 0b07e9ae..b038e2f5 100644 --- a/fission/spec.go +++ b/fission/spec.go @@ -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 } } diff --git a/fission/upgrade.go b/fission/upgrade.go index 81c2a53b..dfdb5a0f 100644 --- a/fission/upgrade.go +++ b/fission/upgrade.go @@ -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 diff --git a/glide.lock b/glide.lock index 1aa3b3a0..fad8953d 100644 --- a/glide.lock +++ b/glide.lock @@ -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 diff --git a/glide.yaml b/glide.yaml index 2aeee8f2..6c096cd3 100644 --- a/glide.yaml +++ b/glide.yaml @@ -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 \ No newline at end of file diff --git a/router/functionHandler.go b/router/functionHandler.go index 6b63469b..36af7e92 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -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) diff --git a/router/router.go b/router/router.go index 7b00af13..c75359d3 100644 --- a/router/router.go +++ b/router/router.go @@ -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() { diff --git a/storagesvc/client/client.go b/storagesvc/client/client.go index a65abf42..ecf590ca 100644 --- a/storagesvc/client/client.go +++ b/storagesvc/client/client.go @@ -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 } diff --git a/storagesvc/client/storagesvc_test.go b/storagesvc/client/storagesvc_test.go index a0f2b684..9649c436 100644 --- a/storagesvc/client/storagesvc_test.go +++ b/storagesvc/client/storagesvc_test.go @@ -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") } diff --git a/storagesvc/storagesvc.go b/storagesvc/storagesvc.go index fb16af63..1c3067e1 100644 --- a/storagesvc/storagesvc.go +++ b/storagesvc/storagesvc.go @@ -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 {