Add method and subpath flags to timetrigger object for triggering a function (#3017)

* Update timetrigger crd and add method and subpath fields in spec.
Update fission-cli to accept user input for method and subpath fields.
Update publisher package to utilize these fields for triggering a function.
Update timer controller to use method and subpath fields for publishing a request.
Add a new test TestPublisherSubpath in pulisher package.

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

* Use kubebuilder default annotation.
Update test for fission-cli timetrigger create, update command to support method and subpath flags.

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

---------

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

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

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

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 {

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{

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),
 		},
 	}
 

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()
 

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

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"}

pkg/fission-cli/flag/key/key.go

@@ -110,6 +110,7 @@ const (
 	TtCron   = "cron"
 	TtFnName = "function"
 	TtRound  = "round"
+	TtMethod = "method"
 
 	MqtName            = resourceName
 	MqtFnName          = "function"

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
+}

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)
 	}
 }
 

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)
 	}
 )

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)
 }

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

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))

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) {