diff --git a/crds/v1/fission.io_timetriggers.yaml b/crds/v1/fission.io_timetriggers.yaml index 9225e71d..3b334189 100644 --- a/crds/v1/fission.io_timetriggers.yaml +++ b/crds/v1/fission.io_timetriggers.yaml @@ -73,6 +73,17 @@ spec: - name - type type: object + method: + default: POST + description: 'HTTP Method for trigger, ex : GET, POST, PUT, DELETE, + HEAD (default: "POST")' + type: string + subpath: + default: / + description: |- + Subpath to trigger a specific route if function + internally supports routing, (default: "/") + type: string required: - cron - functionref diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index 046a9397..11e0ffd8 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -821,6 +821,17 @@ type ( // The reference to function FunctionReference `json:"functionref"` + + // HTTP Method for trigger, ex : GET, POST, PUT, DELETE, HEAD (default: "POST") + // +kubebuilder:default:="POST" + // +optional + Method string `json:"method,omitempty"` + + // Subpath to trigger a specific route if function + // internally supports routing, (default: "/") + // +kubebuilder:default:="/" + // +optional + Subpath string `json:"subpath,omitempty"` } // FailureType refers to the type of failure FailureType string diff --git a/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go b/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go index 53b3a665..0a641f88 100644 --- a/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go +++ b/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go @@ -428,6 +428,8 @@ var map_TimeTriggerSpec = map[string]string{ "": "TimeTriggerSpec invokes the specific function at a time or times specified by a cron string.", "cron": "Cron schedule", "functionref": "The reference to function", + "method": "HTTP Method for trigger, ex : GET, POST, PUT, DELETE, HEAD (default: \"POST\")", + "subpath": "Subpath to trigger a specific route if function internally supports routing, (default: \"/\")", } func (TimeTriggerSpec) SwaggerDoc() map[string]string { diff --git a/pkg/fission-cli/cmd/timetrigger/command.go b/pkg/fission-cli/cmd/timetrigger/command.go index 204cbd77..3b0877fe 100644 --- a/pkg/fission-cli/cmd/timetrigger/command.go +++ b/pkg/fission-cli/cmd/timetrigger/command.go @@ -31,7 +31,11 @@ func Commands() *cobra.Command { } wrapper.SetFlags(createCmd, flag.FlagSet{ Optional: []flag.Flag{flag.TtName, flag.TtFnName, - flag.TtCron, flag.NamespaceFunction, flag.SpecSave, flag.SpecDry}, + flag.TtCron, flag.NamespaceFunction, + flag.TtMethod, flag.FnSubPath, + + flag.SpecSave, flag.SpecDry, + }, }) updateCmd := &cobra.Command{ @@ -42,7 +46,10 @@ func Commands() *cobra.Command { } wrapper.SetFlags(updateCmd, flag.FlagSet{ Required: []flag.Flag{flag.TtName}, - Optional: []flag.Flag{flag.TtFnName, flag.TtCron, flag.NamespaceTrigger}, + Optional: []flag.Flag{flag.TtFnName, flag.TtCron, flag.NamespaceTrigger, + + flag.TtMethod, flag.FnSubPath, + }, }) deleteCmd := &cobra.Command{ diff --git a/pkg/fission-cli/cmd/timetrigger/create.go b/pkg/fission-cli/cmd/timetrigger/create.go index 3dd5e53e..feb6d1e8 100644 --- a/pkg/fission-cli/cmd/timetrigger/create.go +++ b/pkg/fission-cli/cmd/timetrigger/create.go @@ -116,6 +116,8 @@ func (opts *CreateSubCommand) complete(input cli.Input) (err error) { Type: fv1.FunctionReferenceTypeFunctionName, Name: fnName, }, + Method: input.String(flagkey.TtMethod), + Subpath: input.String(flagkey.FnSubPath), }, } diff --git a/pkg/fission-cli/cmd/timetrigger/list.go b/pkg/fission-cli/cmd/timetrigger/list.go index 05001308..3edc65b6 100644 --- a/pkg/fission-cli/cmd/timetrigger/list.go +++ b/pkg/fission-cli/cmd/timetrigger/list.go @@ -54,10 +54,10 @@ func (opts *ListSubCommand) do(input cli.Input) (err error) { w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) - fmt.Fprintf(w, "%v\t%v\t%v\n", "NAME", "CRON", "FUNCTION_NAME") + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n", "NAME", "CRON", "FUNCTION_NAME", "METHOD", "SUBPATH") for _, tt := range tts.Items { - fmt.Fprintf(w, "%v\t%v\t%v\n", - tt.ObjectMeta.Name, tt.Spec.Cron, tt.Spec.FunctionReference.Name) + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n", + tt.ObjectMeta.Name, tt.Spec.Cron, tt.Spec.FunctionReference.Name, tt.Spec.Method, tt.Spec.Subpath) } w.Flush() diff --git a/pkg/fission-cli/cmd/timetrigger/update.go b/pkg/fission-cli/cmd/timetrigger/update.go index ab9300cf..38785c7a 100644 --- a/pkg/fission-cli/cmd/timetrigger/update.go +++ b/pkg/fission-cli/cmd/timetrigger/update.go @@ -77,8 +77,18 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { updated = true } + if input.IsSet(flagkey.TtMethod) { + tt.Spec.Method = input.String(flagkey.TtMethod) + updated = true + } + + if input.IsSet(flagkey.FnSubPath) { + tt.Spec.Subpath = input.String(flagkey.FnSubPath) + updated = true + } + if !updated { - return errors.New("nothing to update. Use --cron or --function") + return errors.New("nothing to update. Use --cron or --function or --method or --subpath") } opts.trigger = tt diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index c1424fd8..7636f965 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -157,6 +157,7 @@ var ( TtCron = Flag{Type: String, Name: flagkey.TtCron, Usage: "Time trigger cron spec with each asterisk representing respectively second, minute, hour, the day of the month, month and day of the week. Also supports readable formats like '@every 5m', '@hourly'"} TtFnName = Flag{Type: String, Name: flagkey.TtFnName, Usage: "Function name"} TtRound = Flag{Type: Int, Name: flagkey.TtRound, Usage: "Get next N rounds of invocation time", DefaultValue: 1} + TtMethod = Flag{Type: String, Name: flagkey.TtMethod, Usage: "HTTP Methods: GET,POST,PUT,DELETE,HEAD."} MqtName = Flag{Type: String, Name: flagkey.MqtName, Usage: "Message queue trigger name"} MqtFnName = Flag{Type: String, Name: flagkey.MqtFnName, Usage: "Function name"} diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go index e2dc9f31..f3347e4d 100644 --- a/pkg/fission-cli/flag/key/key.go +++ b/pkg/fission-cli/flag/key/key.go @@ -110,6 +110,7 @@ const ( TtCron = "cron" TtFnName = "function" TtRound = "round" + TtMethod = "method" MqtName = resourceName MqtFnName = "function" diff --git a/pkg/generated/applyconfiguration/core/v1/timetriggerspec.go b/pkg/generated/applyconfiguration/core/v1/timetriggerspec.go index 0ddc8def..7dfc51b6 100644 --- a/pkg/generated/applyconfiguration/core/v1/timetriggerspec.go +++ b/pkg/generated/applyconfiguration/core/v1/timetriggerspec.go @@ -27,6 +27,8 @@ import ( type TimeTriggerSpecApplyConfiguration struct { Cron *string `json:"cron,omitempty"` *FunctionReferenceApplyConfiguration `json:"functionref,omitempty"` + Method *string `json:"method,omitempty"` + Subpath *string `json:"subpath,omitempty"` } // TimeTriggerSpecApplyConfiguration constructs an declarative configuration of the TimeTriggerSpec type for use with @@ -81,3 +83,19 @@ func (b *TimeTriggerSpecApplyConfiguration) ensureFunctionReferenceApplyConfigur b.FunctionReferenceApplyConfiguration = &FunctionReferenceApplyConfiguration{} } } + +// WithMethod sets the Method field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the Method field is set to the value of the last call. +func (b *TimeTriggerSpecApplyConfiguration) WithMethod(value string) *TimeTriggerSpecApplyConfiguration { + b.Method = &value + return b +} + +// WithSubpath sets the Subpath field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the Subpath field is set to the value of the last call. +func (b *TimeTriggerSpecApplyConfiguration) WithSubpath(value string) *TimeTriggerSpecApplyConfiguration { + b.Subpath = &value + return b +} diff --git a/pkg/kubewatcher/kubewatcher.go b/pkg/kubewatcher/kubewatcher.go index 297801f1..ef55f9e4 100644 --- a/pkg/kubewatcher/kubewatcher.go +++ b/pkg/kubewatcher/kubewatcher.go @@ -22,6 +22,7 @@ import ( "encoding/json" "fmt" "io" + "net/http" "reflect" "strings" "sync/atomic" @@ -268,7 +269,7 @@ func (ws *watchSubscription) eventDispatchLoop(ctx context.Context) { // the triggers can only be created in the same namespace as the function. // so essentially, function namespace = trigger namespace. url := utils.UrlForFunction(ws.watch.Spec.FunctionReference.Name, ws.watch.ObjectMeta.Namespace) - ws.publisher.Publish(ctx, buf.String(), headers, url) + ws.publisher.Publish(ctx, buf.String(), headers, http.MethodPost, url) } } diff --git a/pkg/publisher/publisher.go b/pkg/publisher/publisher.go index 96597614..2d955c48 100644 --- a/pkg/publisher/publisher.go +++ b/pkg/publisher/publisher.go @@ -25,6 +25,6 @@ type ( // Publish a request to a "target". Target's meaning depends on the // publisher: it's a URL in the case of a webhook publisher, or a queue // name in a queue-based publisher such as NATS. - Publish(ctx context.Context, body string, headers map[string]string, target string) + Publish(ctx context.Context, body string, headers map[string]string, method, target string) } ) diff --git a/pkg/publisher/publisher_test.go b/pkg/publisher/publisher_test.go index 3f27a58c..a97eef5c 100644 --- a/pkg/publisher/publisher_test.go +++ b/pkg/publisher/publisher_test.go @@ -29,6 +29,28 @@ func TestPublisher(t *testing.T) { } wp := MakeWebhookPublisher(logger, s.URL) - wp.Publish(ctx, "", map[string]string{"X-Fission-Test": "aaa"}, fnName) + wp.Publish(ctx, "", map[string]string{"X-Fission-Test": "aaa"}, http.MethodPost, fnName) + time.Sleep(time.Second * 1) +} + +func TestPublisherSubpath(t *testing.T) { + subpath := "/api/v1/read" + fnName := "test-fn-subpath" + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/"+fnName+subpath, r.URL.Path) + assert.Equal(t, "aaa", r.Header.Get("X-Fission-Test")) + assert.Contains(t, r.Header, "Traceparent") + })) + + ctx := context.Background() + logger := loggerfactory.GetLogger() + shutdown, err := otelUtils.InitProvider(ctx, logger, fnName) + assert.NoError(t, err) + if shutdown != nil { + defer shutdown(ctx) + } + + wp := MakeWebhookPublisher(logger, s.URL) + wp.Publish(ctx, "", map[string]string{"X-Fission-Test": "aaa"}, http.MethodGet, fnName+subpath) time.Sleep(time.Second * 1) } diff --git a/pkg/publisher/webhookPublisher.go b/pkg/publisher/webhookPublisher.go index 513a9883..ff6c61ea 100644 --- a/pkg/publisher/webhookPublisher.go +++ b/pkg/publisher/webhookPublisher.go @@ -49,6 +49,7 @@ type ( ctx context.Context body string headers map[string]string + method string target string retries int retryDelay time.Duration @@ -72,7 +73,7 @@ func MakeWebhookPublisher(logger *zap.Logger, baseURL string) *WebhookPublisher } // Publish sends a request to the target with payload having given body and headers -func (p *WebhookPublisher) Publish(ctx context.Context, body string, headers map[string]string, target string) { +func (p *WebhookPublisher) Publish(ctx context.Context, body string, headers map[string]string, method, target string) { tracer := otel.Tracer("WebhookPublisher") ctx, span := tracer.Start(ctx, "WebhookPublisher/Publish") defer span.End() @@ -82,6 +83,7 @@ func (p *WebhookPublisher) Publish(ctx context.Context, body string, headers map ctx: ctx, body: body, headers: headers, + method: method, target: target, retries: p.maxRetries, retryDelay: p.retryDelay, @@ -98,7 +100,7 @@ func (p *WebhookPublisher) svc() { func (p *WebhookPublisher) makeHTTPRequest(r *publishRequest) { url := p.baseURL + "/" + strings.TrimPrefix(r.target, "/") - msg := "making HTTP request" + msg := fmt.Sprintf("making HTTP %s request", r.method) level := zap.ErrorLevel fields := []zap.Field{zap.String("url", url), zap.String("type", "publish_request")} @@ -113,7 +115,7 @@ func (p *WebhookPublisher) makeHTTPRequest(r *publishRequest) { buf.WriteString(r.body) // Create request - req, err := http.NewRequest(http.MethodPost, url, &buf) + req, err := http.NewRequest(r.method, url, &buf) if err != nil { fields = append(fields, zap.Error(err)) return diff --git a/pkg/timer/timer.go b/pkg/timer/timer.go index a83999fd..56337bbd 100644 --- a/pkg/timer/timer.go +++ b/pkg/timer/timer.go @@ -57,6 +57,7 @@ func MakeTimer(logger *zap.Logger, publisher publisher.Publisher) *Timer { } func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron { + target := utils.UrlForFunction(t.Spec.FunctionReference.Name, t.Namespace) + t.Spec.Subpath c := cron.New( cron.WithParser( cron.NewParser( @@ -69,7 +70,7 @@ func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron { // with the addition of multi-tenancy, the users can create functions in any namespace. however, // the triggers can only be created in the same namespace as the function. // so essentially, function namespace = trigger namespace. - (*timer.publisher).Publish(context.Background(), "", headers, utils.UrlForFunction(t.Spec.FunctionReference.Name, t.Namespace)) + (*timer.publisher).Publish(context.Background(), "", headers, t.Spec.Method, target) }) c.Start() timer.logger.Info("started cron for time trigger", zap.String("trigger_name", t.Name), zap.String("trigger_namespace", t.Namespace), zap.String("cron", t.Spec.Cron)) diff --git a/test/e2e/cli/cli_test.go b/test/e2e/cli/cli_test.go index 989d9f31..cc4a61bc 100644 --- a/test/e2e/cli/cli_test.go +++ b/test/e2e/cli/cli_test.go @@ -282,6 +282,8 @@ func TestFissionCLI(t *testing.T) { require.Equal(t, "test-tt", tt.Name) require.Equal(t, "test-func", tt.Spec.FunctionReference.Name) require.Equal(t, "@every 1m", tt.Spec.Cron) + require.Equal(t, "POST", tt.Spec.Method) + require.Equal(t, "/", tt.Spec.Subpath) }) t.Run("list", func(t *testing.T) { @@ -295,7 +297,7 @@ func TestFissionCLI(t *testing.T) { }) t.Run("update", func(t *testing.T) { - _, err := cli.ExecCommand(f, ctx, "timetrigger", "update", "--name", "test-tt", "--cron", "@every 2m") + _, err := cli.ExecCommand(f, ctx, "timetrigger", "update", "--name", "test-tt", "--cron", "@every 2m", "--method", "GET", "--subpath", "/api/v1/fetch") require.NoError(t, err) tt, err := fissionClient.CoreV1().TimeTriggers(metav1.NamespaceDefault).Get(ctx, "test-tt", metav1.GetOptions{}) @@ -304,6 +306,8 @@ func TestFissionCLI(t *testing.T) { require.Equal(t, "test-tt", tt.Name) require.Equal(t, "test-func", tt.Spec.FunctionReference.Name) require.Equal(t, "@every 2m", tt.Spec.Cron) + require.Equal(t, "GET", tt.Spec.Method) + require.Equal(t, "/api/v1/fetch", tt.Spec.Subpath) }) t.Run("delete", func(t *testing.T) {