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

Fixed golangci-lint issues: /fission/pkg/router (#1831)

Browse Source
This commit is contained in:
Gaurav Gahlot
2020-11-03 16:15:50 +05:30
committed by GitHub
parent 2f23120a64
commit aaf9f18d93
11 changed files with 77 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/router/analytics.go
+5 -4
View File
@@ -12,21 +12,22 @@ import (
)
type (
// Analytics struct
// Analytics helps exporting usage metrics
Analytics struct {
id string
url string
}
// AnalyticsData struct
// AnalyticsData is the data exported as usage metrics
AnalyticsData struct {
ID string `json:"ID"`
FunctionCallCount uint64 `json:"FunctionCallCount"`
}
)
// MakeAnalytics returns Analytics if url is not empty
// MakeAnalytics returns a new instance of Analytics if url or
// 'ANALYTICS_URL' environment variable is set; nil otherwise
func MakeAnalytics(url string) *Analytics {
if len(url) == 0 {
url = os.Getenv("ANALYTICS_URL")
if len(url) == 0 {
pkg/router/functionHandler.go
+32 -22
View File
@@ -42,7 +42,10 @@ import (
)
const (
FORWARDED = "Forwarded"
// FORWARDED represents the 'Forwarded' request header
FORWARDED = "Forwarded"
// X_FORWARDED_HOST represents the 'X_FORWARDED_HOST' request header
X_FORWARDED_HOST = "X-Forwarded-Host"
)
@@ -54,7 +57,7 @@ type (
function *fv1.Function
httpTrigger *fv1.HTTPTrigger
functionMap map[string]*fv1.Function
fnWeightDistributionList []FunctionWeightDistribution
fnWeightDistributionList []functionWeightDistribution
tsRoundTripperParams *tsRoundTripperParams
isDebugEnv bool
svcAddrUpdateThrottler *throttler.Throttler
@@ -82,13 +85,13 @@ type (
svcAddrRetryCount int
}
// A layer on top of http.DefaultTransport, with retries.
// RetryingRoundTripper is a layer on top of http.DefaultTransport, with retries.
RetryingRoundTripper struct {
logger *zap.Logger
funcHandler *functionHandler
funcTimeout time.Duration
closeContextFunc *context.CancelFunc
serviceUrl *url.URL
serviceURL *url.URL
urlFromCache bool
totalRetry int
}
@@ -184,7 +187,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, err = roundTripper.funcHandler.getServiceEntryFromExecutor()
roundTripper.serviceURL, err = roundTripper.funcHandler.getServiceEntryFromExecutor()
if err != nil {
// We might want a specific error code or header for fission failures as opposed to
// user function bugs.
@@ -210,15 +213,15 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
// }
}
if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr {
defer func(fn *fv1.Function, serviceUrl *url.URL) {
go roundTripper.funcHandler.unTapService(fn, serviceUrl)
}(roundTripper.funcHandler.function, roundTripper.serviceUrl)
defer func(fn *fv1.Function, serviceURL *url.URL) {
go roundTripper.funcHandler.unTapService(fn, serviceURL) //nolint errcheck
}(roundTripper.funcHandler.function, roundTripper.serviceURL)
}
// modify the request to reflect the service url
// this service url comes from executor response
req.URL.Scheme = roundTripper.serviceUrl.Scheme
req.URL.Host = roundTripper.serviceUrl.Host
req.URL.Scheme = roundTripper.serviceURL.Scheme
req.URL.Host = roundTripper.serviceURL.Host
// To keep the function run container simple, it
// doesn't do any routing. In the future if we have
@@ -230,7 +233,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
// Overwrite request host with internal host,
// or request will be blocked in some situations
// (e.g. istio-proxy)
req.Host = roundTripper.serviceUrl.Host
req.Host = roundTripper.serviceURL.Host
}
// over-riding default settings.
@@ -350,11 +353,11 @@ func (roundTripper *RetryingRoundTripper) closeContext() {
}
}
func (fh *functionHandler) tapService(fn *fv1.Function, serviceUrl *url.URL) {
func (fh *functionHandler) tapService(fn *fv1.Function, serviceURL *url.URL) {
if fh.executor == nil {
return
}
fh.executor.TapService(fn.ObjectMeta, fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, serviceUrl)
fh.executor.TapService(fn.ObjectMeta, fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, serviceURL)
}
func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) {
@@ -426,7 +429,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h
// findCeil picks a function from the functionWeightDistribution list based on the
// random number generated. It uses the prefix calculated for the function weights.
func findCeil(randomNumber int, wtDistrList []FunctionWeightDistribution) string {
func findCeil(randomNumber int, wtDistrList []functionWeightDistribution) string {
low := 0
high := len(wtDistrList) - 1
@@ -445,13 +448,12 @@ func findCeil(randomNumber int, wtDistrList []FunctionWeightDistribution) string
if wtDistrList[low].sumPrefix >= randomNumber {
return wtDistrList[low].name
} else {
return ""
}
return ""
}
// picks a function to route to based on a random number generated
func getCanaryBackend(fnMap map[string]*fv1.Function, fnWtDistributionList []FunctionWeightDistribution) *fv1.Function {
func getCanaryBackend(fnMap map[string]*fv1.Function, fnWtDistributionList []functionWeightDistribution) *fv1.Function {
randomNumber := rand.Intn(fnWtDistributionList[len(fnWtDistributionList)-1].sumPrefix + 1)
fnName := findCeil(randomNumber, fnWtDistributionList)
return fnMap[fnName]
@@ -541,15 +543,15 @@ func (fh functionHandler) getServiceEntryFromExecutor() (*url.URL, error) {
}
// parse the address into url
serviceUrl, err := url.Parse(fmt.Sprintf("http://%v", service))
serviceURL, err := url.Parse(fmt.Sprintf("http://%v", service))
if err != nil {
fh.logger.Error("error parsing service url",
zap.Error(err),
zap.String("service_url", serviceUrl.String()))
zap.String("service_url", serviceURL.String()))
return nil, err
}
return serviceUrl, nil
return serviceURL, nil
}
// getProxyErrorHandler returns a reverse proxy error handler
@@ -584,7 +586,15 @@ func (fh functionHandler) getProxyErrorHandler(start time.Time, rrt *RetryingRou
// TODO: return error message that contains traceable UUID back to user. Issue #693
rw.WriteHeader(status)
rw.Write([]byte(msg))
_, err = rw.Write([]byte(msg))
if err != nil {
fh.logger.Error(
"error writing HTTP response",
zap.Error(err),
zap.Any("function", fh.function),
zap.Any("request_header", req.Header),
)
}
}
}
@@ -613,7 +623,7 @@ func (fh functionHandler) collectFunctionMetric(start time.Time, rrt *RetryingRo
// tapService before invoking roundTrip for the serviceUrl
if rrt.urlFromCache {
fh.tapService(fh.function, rrt.serviceUrl)
fh.tapService(fh.function, rrt.serviceURL)
}
fh.logger.Debug("Request complete", zap.String("function", fh.function.ObjectMeta.Name),
pkg/router/functionReferenceResolver.go
+5 -5
View File
@@ -39,7 +39,7 @@ type (
resolveResultType int
FunctionWeightDistribution struct {
functionWeightDistribution struct {
name string
weight int
sumPrefix int
@@ -51,7 +51,7 @@ type (
resolveResult struct {
resolveResultType
functionMap map[string]*fv1.Function
functionWtDistributionList []FunctionWeightDistribution
functionWtDistributionList []functionWeightDistribution
}
// namespacedTriggerReference is just a trigger reference plus a
@@ -112,7 +112,7 @@ func (frr *functionReferenceResolver) resolve(trigger fv1.HTTPTrigger) (*resolve
}
// cache resolve result
frr.refCache.Set(nfr, *rr)
frr.refCache.Set(nfr, *rr) //nolint: errcheck
return rr, nil
}
@@ -149,7 +149,7 @@ func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*re
func (frr *functionReferenceResolver) resolveByFunctionWeights(namespace string, fr *fv1.FunctionReference) (*resolveResult, error) {
functionMap := make(map[string]*fv1.Function)
fnWtDistrList := make([]FunctionWeightDistribution, 0)
fnWtDistrList := make([]functionWeightDistribution, 0)
sumPrefix := 0
for functionName, functionWeight := range fr.FunctionWeights {
@@ -170,7 +170,7 @@ func (frr *functionReferenceResolver) resolveByFunctionWeights(namespace string,
f := obj.(*fv1.Function)
functionMap[f.ObjectMeta.Name] = f
sumPrefix = sumPrefix + functionWeight
fnWtDistrList = append(fnWtDistrList, FunctionWeightDistribution{
fnWtDistrList = append(fnWtDistrList, functionWeightDistribution{
name: functionName,
weight: functionWeight,
sumPrefix: sumPrefix,
pkg/router/functionServiceMap.go
+3 -3
View File
@@ -66,11 +66,11 @@ func (fmap *functionServiceMap) lookup(f *metav1.ObjectMeta) (*url.URL, error) {
return u, nil
}
func (fmap *functionServiceMap) assign(f *metav1.ObjectMeta, serviceUrl *url.URL) {
func (fmap *functionServiceMap) assign(f *metav1.ObjectMeta, serviceURL *url.URL) {
mk := keyFromMetadata(f)
old, err := fmap.cache.Set(*mk, serviceUrl)
old, err := fmap.cache.Set(*mk, serviceURL)
if err != nil {
if *serviceUrl == *(old.(*url.URL)) {
if *serviceURL == *(old.(*url.URL)) {
return
}
fmap.logger.Error("error caching service url for function with a different value", zap.Error(err))
pkg/router/httpTriggers.go
+1
View File
@@ -37,6 +37,7 @@ import (
"github.com/fission/fission/pkg/utils"
)
// HTTPTriggerSet represents an HTTP trigger set
type HTTPTriggerSet struct {
*functionServiceMap
*mutableRouter
pkg/router/mutablemux.go
+1 -1
View File
@@ -34,7 +34,7 @@ type mutableRouter struct {
router atomic.Value // mux.Router
}
func NewMutableRouter(logger *zap.Logger, handler *mux.Router) *mutableRouter {
func newMutableRouter(logger *zap.Logger, handler *mux.Router) *mutableRouter {
mr := mutableRouter{
logger: logger.Named("mutable_router"),
}
pkg/router/mutablemux_test.go
+15 -6
View File
@@ -28,19 +28,28 @@ import (
)
func OldHandler(responseWriter http.ResponseWriter, request *http.Request) {
responseWriter.Write([]byte("old handler"))
_, err := responseWriter.Write([]byte("old handler"))
if err != nil {
log.Fatal(err)
}
}
func NewHandler(responseWriter http.ResponseWriter, request *http.Request) {
responseWriter.Write([]byte("new handler"))
_, err := responseWriter.Write([]byte("new handler"))
if err != nil {
log.Fatal(err)
}
}
func verifyRequest(expectedResponse string) {
targetUrl := "http://localhost:3333"
testRequest(targetUrl, expectedResponse)
targetURL := "http://localhost:3333"
testRequest(targetURL, expectedResponse)
}
func startServer(mr *mutableRouter) {
http.ListenAndServe(":3333", mr)
err := http.ListenAndServe(":3333", mr)
if err != nil {
log.Fatal(err)
}
}
func spamServer(quit chan bool) {
@@ -70,7 +79,7 @@ func TestMutableMux(t *testing.T) {
logger, err := config.Build()
panicIf(err)
mr := NewMutableRouter(logger, muxRouter)
mr := newMutableRouter(logger, muxRouter)
// start http server
log.Print("Start http server")
pkg/router/requesthHeader.go
+1
View File
@@ -26,6 +26,7 @@ import (
)
const (
// HEADERS_FISSION_FUNCTION_PREFIX represents a function prefix request header
HEADERS_FISSION_FUNCTION_PREFIX = "Fission-Function"
)
pkg/router/router.go
+12 -5
View File
@@ -69,9 +69,9 @@ func router(ctx context.Context, logger *zap.Logger, httpTriggerSet *HTTPTrigger
// see issue https://github.com/fission/fission/issues/1317
useEncodedPath, _ := strconv.ParseBool(os.Getenv("USE_ENCODED_PATH"))
if useEncodedPath {
mr = NewMutableRouter(logger, mux.NewRouter().UseEncodedPath())
mr = newMutableRouter(logger, mux.NewRouter().UseEncodedPath())
} else {
mr = NewMutableRouter(logger, mux.NewRouter())
mr = newMutableRouter(logger, mux.NewRouter())
}
httpTriggerSet.subscribeRouter(ctx, mr, resolver)
@@ -83,7 +83,7 @@ func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRat
mr := router(ctx, logger, httpTriggerSet, resolver)
url := fmt.Sprintf(":%v", port)
http.ListenAndServe(url, &ochttp.Handler{
err := http.ListenAndServe(url, &ochttp.Handler{
Handler: mr,
GetStartOptions: func(r *http.Request) trace.StartOptions {
// do not trace router healthz endpoint
@@ -101,6 +101,12 @@ func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRat
}
},
})
if err != nil {
logger.Error(
"HTTP server error",
zap.Error(err),
)
}
}
func serveMetric(logger *zap.Logger) {
@@ -111,7 +117,8 @@ func serveMetric(logger *zap.Logger) {
logger.Fatal("done listening on metrics endpoint", zap.Error(err))
}
func Start(logger *zap.Logger, port int, executorUrl string) {
// Start starts a router
func Start(logger *zap.Logger, port int, executorURL string) {
_ = MakeAnalytics("")
fmap := makeFunctionServiceMap(logger, time.Minute)
@@ -126,7 +133,7 @@ func Start(logger *zap.Logger, port int, executorUrl string) {
logger.Fatal("error waiting for CRDs", zap.Error(err))
}
executor := executorClient.MakeClient(logger, executorUrl)
executor := executorClient.MakeClient(logger, executorURL)
timeoutStr := os.Getenv("ROUTER_ROUND_TRIP_TIMEOUT")
timeout, err := time.ParseDuration(timeoutStr)
pkg/router/util_test.go
+2 -2
View File
@@ -6,8 +6,8 @@ import (
"net/http"
)
func testRequest(targetUrl string, expectedResponse string) {
resp, err := http.Get(targetUrl)
func testRequest(targetURL string, expectedResponse string) {
resp, err := http.Get(targetURL)
if err != nil {
log.Panicf("failed to make get request: %v", err)
}