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

Fission meets OpenTelemetry (#2157)

Browse Source 
* add opentracing section and otelCollectorEndpoint
* initialize OTLP exporter
* pkg/controller: changes for context propagation
* pkg/executor: changes for context propagation
* pkg/fetcher: changes for context propagation
* pkg/router: changes for context propagation
* pkg/storagesvc: changes for context propagation
* set no default value for otel collector endpoint
* update readme and add notes to charts
* move common code to pkg/utils/otel
* adding fn and env as attributes
* don't use otelhttp transport for websocket
* URL ignore with common filter UrlsToIgnore

Note: The web socket example does not work when using OTEL HTTP. Here is an issue related to that on open-telemetry/opentelemetry-js-contrib.

Signed-off-by: Gaurav Gahlot <gauravgahlot0107@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Gaurav Gahlot
2021-08-19 13:00:30 +05:30
committed by GitHub
co-authored by Sanket Sudake
parent a24934f1a0
commit 0cc3ecc2e9
33 changed files with 675 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/controller/api.go
+11 -2
View File
@@ -24,6 +24,7 @@ import (
"strings"
"github.com/gorilla/mux"
"go.opencensus.io/plugin/ochttp"
"go.uber.org/zap"
apiv1 "k8s.io/api/core/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -34,6 +35,7 @@ import (
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/fission-cli/logdb"
"github.com/fission/fission/pkg/info"
"github.com/fission/fission/pkg/utils/otel"
)
var podNamespace string
@@ -263,9 +265,16 @@ func (api *API) GetHandler() http.Handler {
return r
}
func (api *API) Serve(port int) {
func (api *API) Serve(port int, openTracingEnabled bool) {
address := fmt.Sprintf(":%v", port)
api.logger.Info("server started", zap.Int("port", port))
err := http.ListenAndServe(address, api.GetHandler())
var handler http.Handler
if openTracingEnabled {
handler = &ochttp.Handler{Handler: api.GetHandler()}
} else {
handler = otel.GetHandlerWithOTEL(api.GetHandler(), "fission-controller", otel.UrlsToIgnore("/healthz"))
}
err := http.ListenAndServe(address, handler)
api.logger.Fatal("done listening", zap.Error(err))
}
pkg/controller/api_test.go
+1 -1
View File
@@ -374,7 +374,7 @@ func TestMain(m *testing.M) {
panicIf(err)
go Start(logger, 8888, true)
go Start(logger, 8888, true, true)
time.Sleep(5 * time.Second)
pkg/controller/controller.go
+2 -2
View File
@@ -24,7 +24,7 @@ import (
"github.com/fission/fission/pkg/crd"
)
func Start(logger *zap.Logger, port int, unitTestFlag bool) {
func Start(logger *zap.Logger, port int, unitTestFlag bool, openTracingEnabled bool) {
cLogger := logger.Named("controller")
fc, kc, apiExtClient, _, err := crd.MakeFissionClient()
@@ -53,5 +53,5 @@ func Start(logger *zap.Logger, port int, unitTestFlag bool) {
if err != nil {
cLogger.Fatal("failed to start controller", zap.Error(err))
}
api.Serve(port)
api.Serve(port, openTracingEnabled)
}
pkg/executor/api.go
+11 -4
View File
@@ -33,6 +33,7 @@ import (
fv1 "github.com/fission/fission/pkg/apis/core/v1"
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/utils/otel"
)
func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *http.Request) {
@@ -251,11 +252,17 @@ func (executor *Executor) GetHandler() http.Handler {
}
// Serve starts an HTTP server.
func (executor *Executor) Serve(port int) {
func (executor *Executor) Serve(port int, openTracingEnabled bool) {
executor.logger.Info("starting executor API", zap.Int("port", port))
address := fmt.Sprintf(":%v", port)
err := http.ListenAndServe(address, &ochttp.Handler{
Handler: executor.GetHandler(),
})
var handler http.Handler
if openTracingEnabled {
handler = &ochttp.Handler{Handler: executor.GetHandler()}
} else {
handler = otel.GetHandlerWithOTEL(executor.GetHandler(), "fission-executor", otel.UrlsToIgnore("/healthz"))
}
err := http.ListenAndServe(address, handler)
executor.logger.Fatal("done listening", zap.Error(err))
}
pkg/executor/client/client.go
+16 -3
View File
@@ -23,11 +23,14 @@ import (
"io/ioutil"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"time"
"github.com/pkg/errors"
"go.opencensus.io/plugin/ochttp"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.uber.org/zap"
"golang.org/x/net/context/ctxhttp"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -56,14 +59,24 @@ type (
// MakeClient initializes and returns a Client instance.
func MakeClient(logger *zap.Logger, executorURL string) *Client {
openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED"))
if err != nil {
logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err))
}
var hc *http.Client
if openTracingEnabled {
hc = &http.Client{Transport: &ochttp.Transport{}}
} else {
hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
}
c := &Client{
logger: logger.Named("executor_client"),
executorURL: strings.TrimSuffix(executorURL, "/"),
tappedByURL: make(map[string]TapServiceRequest),
requestChan: make(chan TapServiceRequest, 100),
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
httpClient: hc,
}
go c.service()
return c
pkg/executor/executor.go
+2 -2
View File
@@ -255,7 +255,7 @@ func serveMetric(logger *zap.Logger) {
// StartExecutor Starts executor and the executor components such as Poolmgr,
// deploymgr and potential future executor types
func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNamespace string, port int) error {
func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNamespace string, port int, openTracingEnabled bool) error {
fissionClient, kubernetesClient, _, metricsClient, err := crd.MakeFissionClient()
if err != nil {
return errors.Wrap(err, "failed to get kubernetes client")
@@ -345,7 +345,7 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
}
go reaper.CleanupRoleBindings(logger, kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30)
go api.Serve(port)
go api.Serve(port, openTracingEnabled)
go serveMetric(logger)
return nil
pkg/executor/executor_test.go
+1 -1
View File
@@ -173,7 +173,7 @@ func TestExecutor(t *testing.T) {
// create poolmgr
port := 9999
err = StartExecutor(logger, functionNs, "fission-builder", port)
err = StartExecutor(logger, functionNs, "fission-builder", port, true)
if err != nil {
log.Panicf("failed to start poolmgr: %v", err)
}
pkg/fetcher/client/client.go
+18 -5
View File
@@ -6,11 +6,14 @@ import (
"encoding/json"
"io/ioutil"
"net/http"
"os"
"strconv"
"strings"
"time"
"github.com/pkg/errors"
"go.opencensus.io/plugin/ochttp"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.uber.org/zap"
"golang.org/x/net/context/ctxhttp"
@@ -27,12 +30,22 @@ type (
)
func MakeClient(logger *zap.Logger, fetcherUrl string) *Client {
openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED"))
if err != nil {
logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err))
}
var hc *http.Client
if openTracingEnabled {
hc = &http.Client{Transport: &ochttp.Transport{}}
} else {
hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
}
return &Client{
logger: logger.Named("fetcher_client"),
url: strings.TrimSuffix(fetcherUrl, "/"),
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
logger: logger.Named("fetcher_client"),
url: strings.TrimSuffix(fetcherUrl, "/"),
httpClient: hc,
}
}
pkg/fetcher/config/config.go
+10
View File
@@ -282,6 +282,16 @@ func (cfg *Config) addFetcherToPodSpecWithCommand(podSpec *apiv1.PodSpec, mainCo
},
},
},
Env: []apiv1.EnvVar{
{
Name: "OPENTRACING_ENABLED",
Value: os.Getenv("OPENTRACING_ENABLED"),
},
{
Name: "OTEL_COLLECTOR_ENDPOINT",
Value: os.Getenv("OTEL_COLLECTOR_ENDPOINT"),
},
},
}
// Pod is removed from endpoints list for service when it's
pkg/fetcher/fetcher.go
+39 -17
View File
@@ -25,12 +25,14 @@ import (
"net/http"
"os"
"path/filepath"
"strconv"
"time"
"github.com/mholt/archiver"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
"go.opencensus.io/plugin/ochttp"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
k8serr "k8s.io/apimachinery/pkg/api/errors"
@@ -102,6 +104,18 @@ func MakeFetcher(logger *zap.Logger, sharedVolumePath string, sharedSecretPath s
return nil, errors.Wrap(err, "error reading pod namespace from downward volume")
}
openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED"))
if err != nil {
logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err))
}
var hc *http.Client
if openTracingEnabled {
hc = &http.Client{Transport: &ochttp.Transport{}}
} else {
hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
}
return &Fetcher{
logger: fLogger,
sharedVolumePath: sharedVolumePath,
@@ -113,9 +127,7 @@ func MakeFetcher(logger *zap.Logger, sharedVolumePath string, sharedSecretPath s
Name: string(name),
Namespace: string(namespace),
},
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
httpClient: hc,
}, nil
}
@@ -149,6 +161,8 @@ func (fetcher *Fetcher) VersionHandler(w http.ResponseWriter, r *http.Request) {
}
func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if r.Method != "POST" {
http.Error(w, "only POST is supported on this endpoint", http.StatusMethodNotAllowed)
return
@@ -175,14 +189,14 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) {
return
}
pkg, err := fetcher.getPkgInformation(req)
pkg, err := fetcher.getPkgInformation(ctx, req)
if err != nil {
fetcher.logger.Error("error getting package information", zap.Error(err))
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
code, err := fetcher.Fetch(r.Context(), pkg, req)
code, err := fetcher.Fetch(ctx, pkg, req)
if err != nil {
fetcher.logger.Error("error fetching", zap.Error(err))
http.Error(w, err.Error(), code)
@@ -190,7 +204,7 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) {
}
fetcher.logger.Info("checking secrets/cfgmaps")
code, err = fetcher.FetchSecretsAndCfgMaps(req.Secrets, req.ConfigMaps)
code, err = fetcher.FetchSecretsAndCfgMaps(ctx, req.Secrets, req.ConfigMaps)
if err != nil {
fetcher.logger.Error("error fetching secrets and config maps", zap.Error(err))
http.Error(w, err.Error(), code)
@@ -203,6 +217,8 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) {
}
func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if r.Method != "POST" {
http.Error(w, fmt.Sprintf("only POST is supported on this endpoint, %v received", r.Method), http.StatusMethodNotAllowed)
return
@@ -223,7 +239,7 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request
return
}
err = fetcher.SpecializePod(r.Context(), req.FetchReq, req.LoadReq)
err = fetcher.SpecializePod(ctx, req.FetchReq, req.LoadReq)
if err != nil {
fetcher.logger.Error("error specializing pod", zap.Error(err))
http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -353,10 +369,10 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req Functio
// FetchSecretsAndCfgMaps fetches secrets and configmaps specified by user
// It returns the HTTP code and error if any
func (fetcher *Fetcher) FetchSecretsAndCfgMaps(secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) {
func (fetcher *Fetcher) FetchSecretsAndCfgMaps(ctx context.Context, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) {
if len(secrets) > 0 {
for _, secret := range secrets {
data, err := fetcher.kubeClient.CoreV1().Secrets(secret.Namespace).Get(context.TODO(), secret.Name, metav1.GetOptions{})
data, err := fetcher.kubeClient.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{})
if err != nil {
e := "error getting secret from kubeapi"
@@ -400,7 +416,7 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(secrets []fv1.SecretReference, cf
if len(cfgmaps) > 0 {
for _, config := range cfgmaps {
data, err := fetcher.kubeClient.CoreV1().ConfigMaps(config.Namespace).Get(context.TODO(), config.Name, metav1.GetOptions{})
data, err := fetcher.kubeClient.CoreV1().ConfigMaps(config.Namespace).Get(ctx, config.Name, metav1.GetOptions{})
if err != nil {
e := "error getting configmap from kubeapi"
@@ -450,6 +466,8 @@ func (fetcher *Fetcher) FetchSecretsAndCfgMaps(secrets []fv1.SecretReference, cf
}
func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if r.Method != "POST" {
http.Error(w, "only POST is supported on this endpoint", http.StatusMethodNotAllowed)
return
@@ -503,7 +521,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) {
fetcher.logger.Info("starting upload...")
ssClient := storageSvcClient.MakeClient(req.StorageSvcUrl)
fileID, err := ssClient.Upload(r.Context(), dstFilepath, nil)
fileID, err := ssClient.Upload(ctx, dstFilepath, nil)
if err != nil {
e := "error uploading zip file"
fetcher.logger.Error(e, zap.Error(err), zap.String("file", dstFilepath))
@@ -581,10 +599,10 @@ func (fetcher *Fetcher) unarchive(src string, dst string) error {
}
// getPkgInformation gets package information from k8s api server.
func (fetcher *Fetcher) getPkgInformation(req FunctionFetchRequest) (pkg *fv1.Package, err error) {
func (fetcher *Fetcher) getPkgInformation(ctx context.Context, req FunctionFetchRequest) (pkg *fv1.Package, err error) {
maxRetries := 5
for i := 0; i < maxRetries; i++ {
pkg, err = fetcher.fissionClient.CoreV1().Packages(req.Package.Namespace).Get(context.TODO(), req.Package.Name, metav1.GetOptions{})
pkg, err = fetcher.fissionClient.CoreV1().Packages(req.Package.Namespace).Get(ctx, req.Package.Name, metav1.GetOptions{})
if err == nil {
return pkg, nil
}
@@ -618,7 +636,7 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc
fetcher.logger.Info("specialize request done", zap.Duration("elapsed_time", elapsed))
}()
pkg, err := fetcher.getPkgInformation(fetchReq)
pkg, err := fetcher.getPkgInformation(ctx, fetchReq)
if err != nil {
return errors.Wrap(err, "error getting package information")
}
@@ -628,7 +646,7 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc
return errors.Wrap(err, "error fetching deploy package")
}
_, err = fetcher.FetchSecretsAndCfgMaps(fetchReq.Secrets, fetchReq.ConfigMaps)
_, err = fetcher.FetchSecretsAndCfgMaps(ctx, fetchReq.Secrets, fetchReq.ConfigMaps)
if err != nil {
return errors.Wrap(err, "error fetching secrets/configs")
}
@@ -691,6 +709,8 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc
// WsStartHandler is used to generate websocket events in Kubernetes
func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if r.Method != "GET" {
http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed)
return
@@ -700,7 +720,7 @@ func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) {
klog.Errorf("Error creating recorder %s", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(context.TODO(), metav1.ListOptions{
pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(ctx, metav1.ListOptions{
FieldSelector: "metadata.name=" + fetcher.Info.Name,
})
if err != nil {
@@ -721,6 +741,8 @@ func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) {
// WsEndHandler is used to generate inactive events in Kubernetes
func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if r.Method != "GET" {
http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed)
return
@@ -730,7 +752,7 @@ func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) {
klog.Errorf("Error creating recorder %s", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(context.TODO(), metav1.ListOptions{
pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(ctx, metav1.ListOptions{
FieldSelector: "metadata.name=" + fetcher.Info.Name,
})
if err != nil {
pkg/router/functionHandler.go
+31 -6
View File
@@ -32,6 +32,8 @@ import (
"github.com/pkg/errors"
"go.opencensus.io/plugin/ochttp"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
k8stypes "k8s.io/apimachinery/pkg/types"
@@ -40,8 +42,10 @@ import (
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/error/network"
executorClient "github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/router/util"
"github.com/fission/fission/pkg/throttler"
"github.com/fission/fission/pkg/utils"
otelUtils "github.com/fission/fission/pkg/utils/otel"
)
const (
@@ -66,6 +70,7 @@ type (
svcAddrUpdateThrottler *throttler.Throttler
functionTimeoutMap map[k8stypes.UID]int
unTapServiceTimeout time.Duration
openTracingEnabled bool
}
tsRoundTripperParams struct {
@@ -154,10 +159,11 @@ func (w *fakeCloseReadCloser) RealClose() error {
// Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500
// if it returned an error.
func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
ctx := req.Context()
// set the timeout for transport context
roundTripper.addForwardedHostHeader(req)
transport := roundTripper.getDefaultTransport()
ocRoundTripper := &ochttp.Transport{Base: transport}
executingTimeout := roundTripper.funcHandler.tsRoundTripperParams.timeout
@@ -220,7 +226,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
// trying to get new service url from cache/executor.
if retryCounter == 0 {
// get function service url from cache or executor
roundTripper.serviceURL, roundTripper.urlFromCache, err = roundTripper.funcHandler.getServiceEntry()
roundTripper.serviceURL, roundTripper.urlFromCache, err = roundTripper.funcHandler.getServiceEntry(ctx)
if err != nil {
// We might want a specific error code or header for fission failures as opposed to
// user function bugs.
@@ -249,9 +255,9 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
continue
}
if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr {
defer func(fn *fv1.Function, serviceURL *url.URL) {
defer func(ctx context.Context, fn *fv1.Function, serviceURL *url.URL) {
go roundTripper.funcHandler.unTapService(fn, serviceURL) //nolint errcheck
}(roundTripper.funcHandler.function, roundTripper.serviceURL)
}(ctx, roundTripper.funcHandler.function, roundTripper.serviceURL)
}
// modify the request to reflect the service url
@@ -309,8 +315,23 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
dumpReqFunc(newReq)
}
// The otelhttp.NewTransport() does not work with WebSocket.
// This is probably because it modifies the response body.
// Until we find a better solution to handle websocket requests, we will continue to
// use ochttp.Transport(). We check if the request isWebsocketRequest() and use the
// ochttp.Transport() irrespective of open telemetry is enabled or not.
// Related issue: https://github.com/open-telemetry/opentelemetry-js-contrib/issues/12
// forward the request to the function service
resp, err := ocRoundTripper.RoundTrip(newReq)
var resp *http.Response
if roundTripper.funcHandler.openTracingEnabled || util.IsWebsocketRequest(newReq) {
ocRoundTripper := &ochttp.Transport{Base: transport}
resp, err = ocRoundTripper.RoundTrip(newReq)
} else {
otelRoundTripper := otelhttp.NewTransport(transport)
resp, err = otelRoundTripper.RoundTrip(newReq)
}
if err == nil {
// return response back to user
if roundTripper.funcHandler.isDebugEnv {
@@ -494,6 +515,10 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h
rrt.closeContext()
}()
// add attributes to current span
span := trace.SpanFromContext(request.Context())
span.SetAttributes(otelUtils.GetAttributesForFunction(fh.function)...)
proxy.ServeHTTP(responseWriter, request)
}
@@ -657,7 +682,7 @@ func (fh functionHandler) getServiceEntryFromExecutor() (serviceUrl *url.URL, er
}
// getServiceEntryFromExecutor returns service url entry returns from executor
func (fh functionHandler) getServiceEntry() (svcURL *url.URL, cacheHit bool, err error) {
func (fh functionHandler) getServiceEntry(ctx context.Context) (svcURL *url.URL, cacheHit bool, err error) {
if fh.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr {
svcURL, err = fh.getServiceEntryFromExecutor()
return svcURL, false, err
pkg/router/httpTriggers.go
+31 -4
View File
@@ -19,6 +19,8 @@ package router
import (
"context"
"net/http"
"os"
"strconv"
"time"
"github.com/gorilla/mux"
@@ -33,6 +35,7 @@ import (
genInformer "github.com/fission/fission/pkg/generated/informers/externalversions"
"github.com/fission/fission/pkg/throttler"
"github.com/fission/fission/pkg/utils"
"github.com/fission/fission/pkg/utils/otel"
)
// HTTPTriggerSet represents an HTTP trigger set
@@ -110,6 +113,11 @@ func routerHealthHandler(w http.ResponseWriter, r *http.Request) {
func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router {
muxRouter := mux.NewRouter()
openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED"))
if err != nil {
ts.logger.Fatal("error parsing OPENTRACING_ENABLED", zap.Error(err))
}
// HTTP triggers setup by the user
homeHandled := false
for i := range ts.triggers {
@@ -143,6 +151,7 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler,
functionTimeoutMap: fnTimeoutMap,
unTapServiceTimeout: ts.unTapServiceTimeout,
openTracingEnabled: openTracingEnabled,
}
// The functionHandler for HTTP trigger with fn reference type "FunctionReferenceTypeFunctionName",
@@ -157,13 +166,24 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
fh.function = fn
}
}
var ht *mux.Route
var ht *mux.Route
if trigger.Spec.Prefix != nil && *trigger.Spec.Prefix != "" {
ht = muxRouter.PathPrefix(*trigger.Spec.Prefix).HandlerFunc(fh.handler)
if openTracingEnabled {
ht = muxRouter.PathPrefix(*trigger.Spec.Prefix).HandlerFunc(fh.handler)
} else {
handler := otel.GetHandlerWithOTEL(http.HandlerFunc(fh.handler), *trigger.Spec.Prefix)
ht = muxRouter.PathPrefix(*trigger.Spec.Prefix).Handler(handler)
}
} else {
ht = muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler)
if openTracingEnabled {
ht = muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler)
} else {
handler := otel.GetHandlerWithOTEL(http.HandlerFunc(fh.handler), trigger.Spec.RelativeURL)
ht = muxRouter.Handle(trigger.Spec.RelativeURL, handler)
}
}
methods := trigger.Spec.Methods
if len(trigger.Spec.Method) > 0 {
present := false
@@ -212,7 +232,14 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
functionTimeoutMap: fnTimeoutMap,
unTapServiceTimeout: ts.unTapServiceTimeout,
}
muxRouter.PathPrefix(utils.UrlForFunction(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace)).HandlerFunc(fh.handler)
route := utils.UrlForFunction(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace)
if openTracingEnabled {
muxRouter.PathPrefix(route).HandlerFunc(fh.handler)
} else {
otelHandler := otel.GetHandlerWithOTEL(http.HandlerFunc(fh.handler), route)
muxRouter.PathPrefix(route).Handler(otelHandler)
}
}
// Healthz endpoint for the router.
pkg/router/router.go
+36 -27
View File
@@ -53,11 +53,13 @@ import (
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.opencensus.io/plugin/ochttp"
"go.opencensus.io/trace"
"go.opentelemetry.io/otel"
"go.uber.org/zap"
"github.com/fission/fission/pkg/crd"
executorClient "github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/throttler"
otelUtils "github.com/fission/fission/pkg/utils/otel"
)
// request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url
@@ -80,36 +82,38 @@ func router(ctx context.Context, logger *zap.Logger, httpTriggerSet *HTTPTrigger
}
func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRate float64,
httpTriggerSet *HTTPTriggerSet, displayAccessLog bool) {
httpTriggerSet *HTTPTriggerSet, displayAccessLog bool, openTracingEnabled bool) {
mr := router(ctx, logger, httpTriggerSet)
url := fmt.Sprintf(":%v", port)
err := http.ListenAndServe(url, &ochttp.Handler{
Handler: mr,
GetStartOptions: func(r *http.Request) trace.StartOptions {
// do not trace router healthz endpoint
if strings.Compare(r.URL.Path, "/router-healthz") == 0 {
var err error
if openTracingEnabled {
err = http.ListenAndServe(url, &ochttp.Handler{
Handler: mr,
GetStartOptions: func(r *http.Request) trace.StartOptions {
// do not trace router healthz endpoint
if strings.Compare(r.URL.Path, "/router-healthz") == 0 {
return trace.StartOptions{
Sampler: trace.NeverSample(),
}
}
if displayAccessLog {
reqMsg, err := httputil.DumpRequest(r, false)
if err != nil {
logger.Error("error dumping request", zap.Error(err))
}
logger.Info("request dump", zap.String("request", string(reqMsg)))
}
return trace.StartOptions{
Sampler: trace.NeverSample(),
Sampler: trace.ProbabilitySampler(tracingSamplingRate),
}
}
if displayAccessLog {
reqMsg, err := httputil.DumpRequest(r, false)
if err != nil {
logger.Error("error dumping request", zap.Error(err))
}
logger.Info("request dump", zap.String("request", string(reqMsg)))
}
return trace.StartOptions{
Sampler: trace.ProbabilitySampler(tracingSamplingRate),
}
},
})
},
})
} else {
err = http.ListenAndServe(url, otelUtils.GetHandlerWithOTEL(mr, "fission-router", otelUtils.UrlsToIgnore("/router-healthz")))
}
if err != nil {
logger.Error(
"HTTP server error",
zap.Error(err),
)
logger.Error("HTTP server error", zap.Error(err))
}
}
@@ -122,7 +126,7 @@ func serveMetric(logger *zap.Logger) {
}
// Start starts a router
func Start(logger *zap.Logger, port int, executorURL string) {
func Start(logger *zap.Logger, port int, executorURL string, openTracingEnabled bool) {
fmap := makeFunctionServiceMap(logger, time.Minute)
fissionClient, kubeClient, _, _, err := crd.MakeFissionClient()
@@ -252,7 +256,12 @@ func Start(logger *zap.Logger, port int, executorURL string) {
go serveMetric(logger)
logger.Info("starting router", zap.Int("port", port))
ctx, cancel := context.WithCancel(context.Background())
tracer := otel.Tracer("router")
ctx, span := tracer.Start(context.Background(), "router/Start")
defer span.End()
ctxWithCancel, cancel := context.WithCancel(ctx)
defer cancel()
serve(ctx, logger, port, tracingSamplingRate, triggers, displayAccessLog)
serve(ctxWithCancel, logger, port, tracingSamplingRate, triggers, displayAccessLog, openTracingEnabled)
}
pkg/router/util/util.go
+7
View File
@@ -17,6 +17,8 @@ limitations under the License.
package util
import (
"net/http"
v1 "k8s.io/api/networking/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -100,3 +102,8 @@ func GetDeployLabels(trigger *fv1.HTTPTrigger) map[string]string {
"triggerNamespace": trigger.ObjectMeta.Namespace,
}
}
func IsWebsocketRequest(request *http.Request) bool {
return request.Header.Get("Upgrade") == "websocket" &&
request.Header.Get("Connection") == "Upgrade"
}
pkg/storagesvc/client/client.go
+16 -4
View File
@@ -27,10 +27,13 @@ import (
"net/http"
"net/url"
"os"
"strconv"
"strings"
"github.com/pkg/errors"
"go.opencensus.io/plugin/ochttp"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.uber.org/zap"
"golang.org/x/net/context/ctxhttp"
"github.com/fission/fission/pkg/storagesvc"
@@ -45,11 +48,20 @@ type (
// Client creates a storage service client.
func MakeClient(url string) *Client {
openTracingEnabled, err := strconv.ParseBool(os.Getenv("OPENTRACING_ENABLED"))
if err != nil {
fmt.Printf("error parsing OPENTRACING_ENABLED: %v\n", zap.Error(err))
}
var hc *http.Client
if openTracingEnabled {
hc = &http.Client{Transport: &ochttp.Transport{}}
} else {
hc = &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
}
return &Client{
url: strings.TrimSuffix(url, "/") + "/v1",
httpClient: &http.Client{
Transport: &ochttp.Transport{},
},
url: strings.TrimSuffix(url, "/") + "/v1",
httpClient: hc,
}
}
pkg/storagesvc/client/storagesvc_test.go
+2 -3
View File
@@ -95,8 +95,7 @@ func startS3StorageService(endpoint, bucketName, subDir string) {
os.Setenv("STORAGE_S3_REGION", minioRegion)
storage := storagesvc.NewS3Storage()
_ = storagesvc.Start(logger, storage, port)
_ = storagesvc.Start(logger, storage, port, true)
}
func TestS3StorageService(t *testing.T) {
@@ -213,7 +212,7 @@ func TestLocalStorageService(t *testing.T) {
localPath := fmt.Sprintf("/tmp/%v", testID)
_ = os.Mkdir(localPath, os.ModePerm)
storage := storagesvc.NewLocalStorage(localPath)
_ = storagesvc.Start(logger, storage, port)
_ = storagesvc.Start(logger, storage, port, true)
time.Sleep(time.Second)
client := MakeClient(fmt.Sprintf("http://localhost:%v/", port))
pkg/storagesvc/storagesvc.go
+13 -8
View File
@@ -29,6 +29,8 @@ import (
"github.com/pkg/errors"
"go.opencensus.io/plugin/ochttp"
"go.uber.org/zap"
"github.com/fission/fission/pkg/utils/otel"
)
type (
@@ -132,7 +134,6 @@ func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request)
zap.String("filename", handler.Filename),
)
}
}
func (ss *StorageService) getIdFromRequest(r *http.Request) (string, error) {
@@ -199,7 +200,7 @@ func MakeStorageService(logger *zap.Logger, storageClient *StowClient, port int)
}
}
func (ss *StorageService) Start(port int) {
func (ss *StorageService) Start(port int, openTracingEnabled bool) {
r := mux.NewRouter()
r.HandleFunc("/v1/archive", ss.uploadHandler).Methods("POST")
r.HandleFunc("/v1/archive", ss.downloadHandler).Methods("GET")
@@ -208,15 +209,19 @@ func (ss *StorageService) Start(port int) {
address := fmt.Sprintf(":%v", port)
err := http.ListenAndServe(address, &ochttp.Handler{
Handler: r,
})
var err error
if openTracingEnabled {
err = http.ListenAndServe(address, &ochttp.Handler{
Handler: r,
})
} else {
err = http.ListenAndServe(address, otel.GetHandlerWithOTEL(r, "fission-storagesvc", otel.UrlsToIgnore("/healthz")))
}
ss.logger.Fatal("done listening", zap.Error(err))
}
// Start runs storage service
func Start(logger *zap.Logger, storage Storage, port int) error {
func Start(logger *zap.Logger, storage Storage, port int, openTracingEnabled bool) error {
enablePruner := true
// create a storage client
storageClient, err := MakeStowClient(logger, storage)
@@ -226,7 +231,7 @@ func Start(logger *zap.Logger, storage Storage, port int) error {
// create http handlers
storageService := MakeStorageService(logger, storageClient, port)
go storageService.Start(port)
go storageService.Start(port, openTracingEnabled)
// enablePruner prevents storagesvc unit test from needing to talk to kubernetes
if enablePruner {
pkg/utils/otel/attributes.go
+24
View File
@@ -0,0 +1,24 @@
package otel
import (
"go.opentelemetry.io/otel/attribute"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
)
/* GetAttributesForFunction returns a set of attributes for a function. Attributes returned:
function-name
function-namespace
environment-name
environment-namespace
These attributes are tags that can be used to filter traces.
*/
func GetAttributesForFunction(fn *fv1.Function) []attribute.KeyValue {
return []attribute.KeyValue{
{Key: "function-name", Value: attribute.StringValue(fn.Name)},
{Key: "function-namespace", Value: attribute.StringValue(fn.Namespace)},
{Key: "environment-name", Value: attribute.StringValue(fn.Spec.Environment.Name)},
{Key: "environment-namespace", Value: attribute.StringValue(fn.Spec.Environment.Namespace)},
}
}
pkg/utils/otel/handler.go
+31
View File
@@ -0,0 +1,31 @@
package otel
import (
"net/http"
"strings"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)
func UrlsToIgnore(ignoreEndpoints ...string) func(r *http.Request) bool {
return func(r *http.Request) bool {
for _, ignore := range ignoreEndpoints {
if strings.HasPrefix(r.URL.Path, ignore) {
return false
}
}
return true
}
}
func GetHandlerWithOTEL(h http.Handler, name string, filter ...otelhttp.Filter) http.Handler {
opts := []otelhttp.Option{
otelhttp.WithMessageEvents(otelhttp.ReadEvents, otelhttp.WriteEvents),
}
for _, f := range filter {
opts = append(opts, otelhttp.WithFilter(f))
}
return otelhttp.NewHandler(h, name, opts...)
}
pkg/utils/otel/provider.go
+61
View File
@@ -0,0 +1,61 @@
package otel
import (
"context"
"os"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
"go.uber.org/zap"
"google.golang.org/grpc"
)
// Initializes an OTLP exporter, and configures the corresponding trace and metric providers.
func InitProvider(logger *zap.Logger, serviceName string) (func(), error) {
collectorEndpoint := os.Getenv("OTEL_COLLECTOR_ENDPOINT")
if collectorEndpoint == "" {
logger.Info("skipping trace exporter registration")
return nil, nil
}
ctx := context.Background()
res, err := resource.New(ctx,
resource.WithAttributes(
semconv.ServiceNameKey.String(serviceName),
),
)
if err != nil {
return nil, err
}
traceExporter, err := otlptracegrpc.New(ctx,
otlptracegrpc.WithInsecure(),
otlptracegrpc.WithEndpoint(collectorEndpoint),
otlptracegrpc.WithDialOption(grpc.WithBlock()),
)
if err != nil {
return nil, err
}
bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
tracerProvider := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sdktrace.AlwaysSample()),
sdktrace.WithResource(res),
sdktrace.WithSpanProcessor(bsp),
)
otel.SetTracerProvider(tracerProvider)
otel.SetTextMapPropagator(propagation.TraceContext{})
// Shutdown will flush any remaining spans and shut down the exporter.
return func() {
err := tracerProvider.Shutdown(ctx)
if err != nil {
logger.Fatal("error shutting down trace provider", zap.Error(err))
}
}, nil
}
pkg/utils/tracing/tracing.go
+45
View File
@@ -0,0 +1,45 @@
package tracing
import (
"os"
"strconv"
"strings"
"contrib.go.opencensus.io/exporter/jaeger"
"go.opencensus.io/trace"
"go.uber.org/zap"
)
func RegisterTraceExporter(logger *zap.Logger, collectorEndpoint, serviceName string) error {
if len(collectorEndpoint) == 0 {
logger.Info("skipping trace exporter registration")
return nil
}
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)
if strings.EqualFold(serviceName, "Fission-Fetcher") {
trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
} else {
samplingRate, err := strconv.ParseFloat(os.Getenv("TRACING_SAMPLING_RATE"), 32)
if err != nil {
return err
}
trace.ApplyConfig(trace.Config{DefaultSampler: trace.ProbabilitySampler(samplingRate)})
}
return nil
}