diff --git a/Makefile b/Makefile index 62dcdb9c..3a2dac16 100644 --- a/Makefile +++ b/Makefile @@ -99,7 +99,7 @@ generate-cli-docs: go run tools/cmd-docs/main.go -o "../fission.io/content/en/docs/reference/fission-cli" install-crd-ref-docs: - go install github.com/elastic/crd-ref-docs@v0.0.9 + go install github.com/elastic/crd-ref-docs@v0.0.10 generate-crd-ref-docs: install-crd-ref-docs # crd-ref-docs: https://github.com/elastic/crd-ref-docs diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index 1cc2b220..ed9ab734 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -32,6 +32,7 @@ import ( "github.com/fission/fission/pkg/canaryconfigmgr" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor" + eclient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/info" "github.com/fission/fission/pkg/kubewatcher" functionLogger "github.com/fission/fission/pkg/logger" @@ -55,7 +56,7 @@ func runCanaryConfigServer(ctx context.Context, clientGen crd.ClientGeneratorInt } func runRouter(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, port int, executorUrl string) error { - return router.Start(ctx, clientGen, logger, port, executorUrl) + return router.Start(ctx, clientGen, logger, port, eclient.MakeClient(logger, executorUrl)) } func runExecutor(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, port int) error { diff --git a/cmd/fission-bundle/mqtrigger/mqtrigger.go b/cmd/fission-bundle/mqtrigger/mqtrigger.go index 67f26b3c..c69da1a1 100644 --- a/cmd/fission-bundle/mqtrigger/mqtrigger.go +++ b/cmd/fission-bundle/mqtrigger/mqtrigger.go @@ -40,7 +40,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * return errors.Wrap(err, "failed to get fission client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/builder/client/client.go b/pkg/builder/client/client.go index 448397a1..5977f6d5 100644 --- a/pkg/builder/client/client.go +++ b/pkg/builder/client/client.go @@ -35,25 +35,29 @@ import ( ) type ( - Client struct { + ClientInterface interface { + Build(context.Context, *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) + } + + client struct { logger *zap.Logger url string httpClient *retryablehttp.Client } ) -func MakeClient(logger *zap.Logger, builderUrl string) *Client { +func MakeClient(logger *zap.Logger, builderUrl string) ClientInterface { hc := retryablehttp.NewClient() hc.ErrorHandler = retryablehttp.PassthroughErrorHandler hc.HTTPClient.Transport = otelhttp.NewTransport(hc.HTTPClient.Transport) - return &Client{ + return &client{ logger: logger.Named("builder_client"), url: strings.TrimSuffix(builderUrl, "/"), httpClient: hc, } } -func (c *Client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) { +func (c *client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) { logger := otelUtils.LoggerWithTraceID(ctx, c.logger) body, err := json.Marshal(req) diff --git a/pkg/buildermgr/buildermgr.go b/pkg/buildermgr/buildermgr.go index 172dfa9b..cd7a5258 100644 --- a/pkg/buildermgr/buildermgr.go +++ b/pkg/buildermgr/buildermgr.go @@ -44,7 +44,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * return errors.Wrap(err, "failed to get kubernetes client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/crd/client.go b/pkg/crd/client.go index dcc1f52d..462ef1ca 100644 --- a/pkg/crd/client.go +++ b/pkg/crd/client.go @@ -18,7 +18,7 @@ package crd import ( "context" - "errors" + "fmt" "time" "go.uber.org/zap" @@ -110,13 +110,13 @@ func NewClientGeneratorWithRestConfig(restConfig *rest.Config) *ClientGenerator return &ClientGenerator{restConfig: restConfig} } -// WaitForCRDs does a timeout to check if CRDs have been installed -func WaitForCRDs(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface) error { - logger.Info("Waiting for CRDs to be installed") +// WaitForFunctionCRDs does a timeout to check if CRDs have been installed +func WaitForFunctionCRDs(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface) error { defaultNs := utils.DefaultNSResolver().DefaultNamespace if defaultNs == "" { defaultNs = metav1.NamespaceDefault } + logger.Info("Checking function CRD access", zap.String("namespace", defaultNs), zap.String("timeout", "30s")) start := time.Now() for { fi := fissionClient.CoreV1().Functions(defaultNs) @@ -128,7 +128,7 @@ func WaitForCRDs(ctx context.Context, logger *zap.Logger, fissionClient versione } if time.Since(start) > 30*time.Second { - return errors.New("timeout waiting for CRDs") + return fmt.Errorf("timeout waiting for function CRD access") } } } diff --git a/pkg/executor/client/client.go b/pkg/executor/client/client.go index 99d66f47..c32fcc48 100644 --- a/pkg/executor/client/client.go +++ b/pkg/executor/client/client.go @@ -36,8 +36,14 @@ import ( ) type ( - // Client is wrapper on a HTTP client. - Client struct { + // ClientInterface is the interface for executor client. + ClientInterface interface { + GetServiceForFunction(ctx context.Context, fn *fv1.Function) (string, error) + TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL url.URL) + UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) error + } + // client is wrapper on a HTTP client. + client struct { logger *zap.Logger executorURL string tappedByURL map[string]TapServiceRequest @@ -54,10 +60,10 @@ type ( ) // MakeClient initializes and returns a Client instance. -func MakeClient(logger *zap.Logger, executorURL string) *Client { +func MakeClient(logger *zap.Logger, executorURL string) ClientInterface { hc := retryablehttp.NewClient() hc.HTTPClient.Transport = otelhttp.NewTransport(hc.HTTPClient.Transport) - c := &Client{ + c := &client{ logger: logger.Named("executor_client"), executorURL: strings.TrimSuffix(executorURL, "/"), tappedByURL: make(map[string]TapServiceRequest), @@ -69,7 +75,7 @@ func MakeClient(logger *zap.Logger, executorURL string) *Client { } // GetServiceForFunction returns the service name for a given function. -func (c *Client) GetServiceForFunction(ctx context.Context, fn *fv1.Function) (string, error) { +func (c *client) GetServiceForFunction(ctx context.Context, fn *fv1.Function) (string, error) { executorURL := c.executorURL + "/v2/getServiceForFunction" body, err := json.Marshal(fn) @@ -102,7 +108,7 @@ func (c *Client) GetServiceForFunction(ctx context.Context, fn *fv1.Function) (s } // UnTapService sends a request to /v2/unTapService. -func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) error { +func (c *client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL *url.URL) error { url := c.executorURL + "/v2/unTapService" tapSvc := TapServiceRequest{ FnMetadata: fnMeta, @@ -133,7 +139,7 @@ func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, exe return nil } -func (c *Client) service() { +func (c *client) service() { ticker := time.NewTicker(time.Second * 5) for { select { @@ -163,7 +169,7 @@ func (c *Client) service() { } // TapService sends a TapServiceRequest over the request channel. -func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL url.URL) { +func (c *client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceURL url.URL) { c.requestChan <- TapServiceRequest{ FnMetadata: metav1.ObjectMeta{ Name: fnMeta.Name, @@ -178,7 +184,7 @@ func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorT } } -func (c *Client) _tapService(ctx context.Context, tapSvcReqs []TapServiceRequest) error { +func (c *client) _tapService(ctx context.Context, tapSvcReqs []TapServiceRequest) error { executorURL := c.executorURL + "/v2/tapServices" body, err := json.Marshal(tapSvcReqs) diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index e2119bad..0559f2a1 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -266,7 +266,7 @@ func StartExecutor(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger.Error("error making the metrics client", zap.Error(err)) } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/fetcher/client/client.go b/pkg/fetcher/client/client.go index 00e26f23..7a9849fd 100644 --- a/pkg/fetcher/client/client.go +++ b/pkg/fetcher/client/client.go @@ -19,45 +19,50 @@ import ( ) type ( - Client struct { + ClientInterface interface { + Specialize(context.Context, *fetcher.FunctionSpecializeRequest) error + Fetch(context.Context, *fetcher.FunctionFetchRequest) error + Upload(context.Context, *fetcher.ArchiveUploadRequest) (*fetcher.ArchiveUploadResponse, error) + } + client struct { logger *zap.Logger url string httpClient *http.Client } ) -func MakeClient(logger *zap.Logger, fetcherUrl string) *Client { +func MakeClient(logger *zap.Logger, fetcherUrl string) ClientInterface { hc := &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} - return &Client{ + return &client{ logger: logger.Named("fetcher_client"), url: strings.TrimSuffix(fetcherUrl, "/"), httpClient: hc, } } -func (c *Client) getSpecializeUrl() string { +func (c *client) getSpecializeUrl() string { return c.url + "/specialize" } -func (c *Client) getFetchUrl() string { +func (c *client) getFetchUrl() string { return c.url + "/fetch" } -func (c *Client) getUploadUrl() string { +func (c *client) getUploadUrl() string { return c.url + "/upload" } -func (c *Client) Specialize(ctx context.Context, req *fetcher.FunctionSpecializeRequest) error { +func (c *client) Specialize(ctx context.Context, req *fetcher.FunctionSpecializeRequest) error { _, err := sendRequest(c.logger, ctx, c.httpClient, req, c.getSpecializeUrl()) return err } -func (c *Client) Fetch(ctx context.Context, fr *fetcher.FunctionFetchRequest) error { +func (c *client) Fetch(ctx context.Context, fr *fetcher.FunctionFetchRequest) error { _, err := sendRequest(c.logger, ctx, c.httpClient, fr, c.getFetchUrl()) return err } -func (c *Client) Upload(ctx context.Context, fr *fetcher.ArchiveUploadRequest) (*fetcher.ArchiveUploadResponse, error) { +func (c *client) Upload(ctx context.Context, fr *fetcher.ArchiveUploadRequest) (*fetcher.ArchiveUploadResponse, error) { body, err := sendRequest(c.logger, ctx, c.httpClient, fr, c.getUploadUrl()) if err != nil { return nil, err diff --git a/pkg/kubewatcher/main.go b/pkg/kubewatcher/main.go index 8c19fe03..e9e617d0 100644 --- a/pkg/kubewatcher/main.go +++ b/pkg/kubewatcher/main.go @@ -36,7 +36,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * return errors.Wrap(err, "failed to get kubernetes client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/mqtrigger/scalermanager.go b/pkg/mqtrigger/scalermanager.go index a0d1d9c6..05ec89c2 100644 --- a/pkg/mqtrigger/scalermanager.go +++ b/pkg/mqtrigger/scalermanager.go @@ -154,7 +154,7 @@ func StartScalerManager(ctx context.Context, clientGen crd.ClientGeneratorInterf return errors.Wrap(err, "failed to get dynamic client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index c783d333..8714ec61 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -38,7 +38,7 @@ import ( "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/error/network" - executorClient "github.com/fission/fission/pkg/executor/client" + eclient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" otelUtils "github.com/fission/fission/pkg/utils/otel" @@ -56,7 +56,7 @@ type ( functionHandler struct { logger *zap.Logger fmap *functionServiceMap - executor *executorClient.Client + executor eclient.ClientInterface function *fv1.Function httpTrigger *fv1.HTTPTrigger functionMap map[string]*fv1.Function diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index cf5cb64d..2bbb13dd 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -32,7 +32,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" - executorClient "github.com/fission/fission/pkg/executor/client" + eclient "github.com/fission/fission/pkg/executor/client" config "github.com/fission/fission/pkg/featureconfig" "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/info" @@ -49,7 +49,7 @@ type HTTPTriggerSet struct { logger *zap.Logger fissionClient versioned.Interface kubeClient kubernetes.Interface - executor *executorClient.Client + executor eclient.ClientInterface resolver *functionReferenceResolver triggers []fv1.HTTPTrigger triggerInformer map[string]k8sCache.SharedIndexInformer @@ -64,7 +64,7 @@ type HTTPTriggerSet struct { } func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionClient versioned.Interface, - kubeClient kubernetes.Interface, executor *executorClient.Client, params *tsRoundTripperParams, isDebugEnv bool, unTapServiceTimeout time.Duration, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, error) { + kubeClient kubernetes.Interface, executor eclient.ClientInterface, params *tsRoundTripperParams, isDebugEnv bool, unTapServiceTimeout time.Duration, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, error) { httpTriggerSet := &HTTPTriggerSet{ logger: logger.Named("http_trigger_set"), diff --git a/pkg/router/router.go b/pkg/router/router.go index 50ac8083..ed4ef976 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -52,7 +52,7 @@ import ( "go.uber.org/zap" "github.com/fission/fission/pkg/crd" - executorClient "github.com/fission/fission/pkg/executor/client" + eclient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils/httpserver" "github.com/fission/fission/pkg/utils/metrics" @@ -98,7 +98,7 @@ func serve(ctx context.Context, logger *zap.Logger, port int, } // Start starts a router -func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, port int, executorURL string) error { +func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger *zap.Logger, port int, executor eclient.ClientInterface) error { fmap := makeFunctionServiceMap(logger, time.Minute) fissionClient, err := clientGen.GetFissionClient() @@ -110,13 +110,11 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * return errors.Wrap(err, "error making the kube client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } - executor := executorClient.MakeClient(logger, executorURL) - timeoutStr := os.Getenv("ROUTER_ROUND_TRIP_TIMEOUT") timeout, err := time.ParseDuration(timeoutStr) if err != nil { diff --git a/pkg/storagesvc/client/client.go b/pkg/storagesvc/client/client.go index 3eff1c4d..7c9321f6 100644 --- a/pkg/storagesvc/client/client.go +++ b/pkg/storagesvc/client/client.go @@ -36,16 +36,24 @@ import ( ) type ( - Client struct { + ClientInterface interface { + Upload(ctx context.Context, filePath string, metadata *map[string]string) (string, error) + GetUrl(id string) string + List(ctx context.Context) ([]string, error) + Download(ctx context.Context, id string, filePath string) error + GetFile(ctx context.Context, id string) (*http.Response, error) + Delete(ctx context.Context, id string) error + } + client struct { url string httpClient *http.Client } ) // Client creates a storage service client. -func MakeClient(url string) *Client { +func MakeClient(url string) ClientInterface { hc := &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)} - return &Client{ + return &client{ url: strings.TrimSuffix(url, "/") + "/v1", httpClient: hc, } @@ -54,7 +62,7 @@ func MakeClient(url string) *Client { // 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(ctx context.Context, 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 @@ -112,11 +120,11 @@ func (c *Client) Upload(ctx context.Context, filePath string, metadata *map[stri } // GetUrl returns an HTTP URL that can be used to download the file pointed to by ID -func (c *Client) GetUrl(id string) string { +func (c *client) GetUrl(id string) string { return fmt.Sprintf("%v/archive?id=%v", c.url, url.PathEscape(id)) } -func (c *Client) List(ctx context.Context) ([]string, error) { +func (c *client) List(ctx context.Context) ([]string, error) { req, err := http.NewRequest(http.MethodGet, c.url+"/archive", nil) if err != nil { return []string{}, err @@ -145,7 +153,7 @@ func (c *Client) List(ctx context.Context) ([]string, error) { // Download fetches the file identified by ID to the local file path. // filePath must not exist. -func (c *Client) Download(ctx context.Context, id string, filePath string) error { +func (c *client) Download(ctx context.Context, id string, filePath string) error { // url for id url := c.GetUrl(id) @@ -187,7 +195,7 @@ func (c *Client) Download(ctx context.Context, id string, filePath string) error // Download fetches the file identified by ID to the local file path. // filePath must not exist. -func (c *Client) GetFile(ctx context.Context, id string) (resp *http.Response, err error) { +func (c *client) GetFile(ctx context.Context, id string) (resp *http.Response, err error) { // url for id url := c.GetUrl(id) @@ -200,7 +208,7 @@ func (c *Client) GetFile(ctx context.Context, id string) (resp *http.Response, e return resp, err } -func (c *Client) Delete(ctx context.Context, 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) diff --git a/pkg/timer/main.go b/pkg/timer/main.go index ff55052d..c6b4463c 100644 --- a/pkg/timer/main.go +++ b/pkg/timer/main.go @@ -32,7 +32,7 @@ func Start(ctx context.Context, clientGen crd.ClientGeneratorInterface, logger * return errors.Wrap(err, "failed to get fission client") } - err = crd.WaitForCRDs(ctx, logger, fissionClient) + err = crd.WaitForFunctionCRDs(ctx, logger, fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/test/e2e/framework/services/services.go b/test/e2e/framework/services/services.go index ca1971f1..1387a038 100644 --- a/test/e2e/framework/services/services.go +++ b/test/e2e/framework/services/services.go @@ -7,6 +7,7 @@ import ( "github.com/fission/fission/pkg/buildermgr" "github.com/fission/fission/pkg/executor" + eclient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/router" "github.com/fission/fission/pkg/storagesvc" "github.com/fission/fission/pkg/utils" @@ -93,7 +94,8 @@ func StartServices(ctx context.Context, f *framework.Framework) error { if err != nil { return fmt.Errorf("error toggling metric address: %v", err) } - err = router.Start(ctx, f.ClientGen(), f.Logger(), routerPort, fmt.Sprintf("http://localhost:%d", executorPort)) + executor := eclient.MakeClient(f.Logger(), fmt.Sprintf("http://localhost:%d", executorPort)) + err = router.Start(ctx, f.ClientGen(), f.Logger(), routerPort, executor) if err != nil { return fmt.Errorf("error starting router: %v", err) }