Merge pull request #176 from xiekeyang/config-namespace-env

Parse and Resolve the Namespace of Fission function

examples/misc/deploy.sh

@@ -6,9 +6,11 @@ set -x
 
 fission env create --name binary --image fission/binary-env
 fission fn create --name dump --env binary --deploy ./dump.sh
+fission fn create --name multinsdump --fns fission-function --env binary --deploy ./dump.sh
 
 # Deploy workflows
 fission fn create --name inputs --env workflow --src ./inputs.wf.yaml
 fission fn create --name fibonacci --env workflow --src ./fibonacci.wf.yaml
 fission fn create --name sleepalot --env workflow --src ./sleepalot.wf.yaml
-fission fn create --name fission-inputs --env workflow --src ./fission-inputs.wf.yaml
+fission fn create --name fission-inputs --env workflow --src ./fission-inputs.wf.yaml
+fission fn create --name multinamespace --fns fission-function --env workflow --src ./multinamespace.wf.yaml

examples/misc/multinamespace.wf.yaml

@@ -0,0 +1,10 @@
+# multinamespace shows you how to access and refere a fission function in task under specific namespace.
+#
+# Usage example: fission fn test --name multinamespace --fns fission-function -b 'Hello World'
+apiVersion: 1
+output: multinamespace
+
+tasks:
+  multinamespace:
+    run: fission://fission-function/multinsdump
+    inputs: "{ $.Invocation.Inputs }"

pkg/fnenv/fission/resolver.go

@@ -1,6 +1,7 @@
 package fission
 
 import (
+	"github.com/fission/fission-workflows/pkg/types"
 	"github.com/fission/fission/controller/client"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -15,18 +16,18 @@ func NewResolver(controller *client.Client) *Resolver {
 	return &Resolver{controller}
 }
 
-func (re *Resolver) Resolve(fnName string) (string, error) {
+func (re *Resolver) Resolve(ref types.FnRef) (string, error) {
 	// Currently we just use the controller API to check if the function exists.
-	log.Infof("Resolving function: %s", fnName)
+	log.Infof("Resolving function: %s", ref.ID)
 	_, err := re.controller.FunctionGet(&metav1.ObjectMeta{
-		Name:      fnName,
-		Namespace: metav1.NamespaceDefault,
+		Name:      ref.ID,
+		Namespace: ref.Namespace,
 	})
 	if err != nil {
 		return "", err
 	}
-	id := fnName
+	id := ref.ID
 
-	log.Infof("Resolved fission function %s to %s", fnName, id)
+	log.Infof("Resolved fission function %s to %s", ref.ID, id)
 	return id, nil
 }

pkg/fnenv/fission/runtime.go

@@ -156,5 +156,9 @@ func createFunctionMeta(fn types.FnRef) *metav1.ObjectMeta {
 }
 
 func (fe *FunctionEnv) createRouterURL(fn types.FnRef) string {
-	return fe.routerURL + "/fission-function/" + fn.ID
+	if fn.Namespace == metav1.NamespaceDefault {
+		return fe.routerURL + "/fission-function/" + fn.ID
+	}
+
+	return fe.routerURL + "/fission-function/" + fn.Namespace + "/" + fn.ID
 }

pkg/fnenv/fnenv.go

@@ -105,5 +105,5 @@ type RuntimeResolver interface {
 
 	// ResolveTask resolved an ambiguous target function name to a unique identifier of a function within the scope
 	// of a runtime.
-	Resolve(targetFn string) (string, error)
+	Resolve(ref types.FnRef) (string, error)
 }

pkg/fnenv/mock/mock.go

@@ -145,10 +145,10 @@ type Resolver struct {
 	FnNameIDs map[string]string
 }
 
-func (mf *Resolver) Resolve(fnName string) (string, error) {
-	fnID, ok := mf.FnNameIDs[fnName]
+func (mf *Resolver) Resolve(ref types.FnRef) (string, error) {
+	fnID, ok := mf.FnNameIDs[ref.ID]
 	if !ok {
-		return "", fmt.Errorf("could not resolve function '%s' using resolve-map '%v'", fnName, mf.FnNameIDs)
+		return "", fmt.Errorf("could not resolve function '%s' using resolve-map '%v'", ref.ID, mf.FnNameIDs)
 	}
 
 	return fnID, nil

pkg/fnenv/native/native.go

@@ -83,13 +83,13 @@ func (fe *FunctionEnv) Invoke(spec *types.TaskInvocationSpec) (*types.TaskInvoca
 	}, nil
 }
 
-func (fe *FunctionEnv) Resolve(fnName string) (string, error) {
-	_, ok := fe.fns[fnName]
+func (fe *FunctionEnv) Resolve(ref types.FnRef) (string, error) {
+	_, ok := fe.fns[ref.ID]
 	if !ok {
-		return "", fmt.Errorf("could not resolve internal function '%s'", fnName)
+		return "", fmt.Errorf("could not resolve internal function '%s'", ref.ID)
 	}
-	log.WithField("name", fnName).WithField("uid", fnName).Debug("Resolved internal function")
-	return fnName, nil
+	log.WithField("name", ref.ID).WithField("uid", ref.ID).Debug("Resolved internal function")
+	return ref.ID, nil
 }
 
 func (fe *FunctionEnv) RegisterFn(name string, fn InternalFunction) {

pkg/fnenv/resolver.go

@@ -52,17 +52,11 @@ func NewMetaResolver(client map[string]RuntimeResolver) *MetaResolver {
 func (ps *MetaResolver) Resolve(targetFn string) (types.FnRef, error) {
 	ref, err := types.ParseFnRef(targetFn)
 	if err != nil {
-		if err == types.ErrNoRuntime {
-			ref = types.FnRef{
-				ID: targetFn,
-			}
-		} else {
-			return types.FnRef{}, err
-		}
+		return types.FnRef{}, err
 	}
 
 	if ref.Runtime != "" {
-		return ps.resolveForRuntime(ref.ID, ref.Runtime)
+		return ps.resolveForRuntime(ref)
 	}
 
 	waitFor := len(ps.clients)
@@ -73,7 +67,7 @@ func (ps *MetaResolver) Resolve(targetFn string) (types.FnRef, error) {
 	var lastErr error
 	for cName := range ps.clients {
 		go func(cName string) {
-			def, err := ps.resolveForRuntime(ref.ID, cName)
+			def, err := ps.resolveForRuntime(types.FnRef{Runtime: cName, Namespace: ref.Namespace, ID: ref.ID})
 			if err != nil {
 				logrus.WithFields(logrus.Fields{
 					"err":     err,
@@ -99,20 +93,21 @@ func (ps *MetaResolver) Resolve(targetFn string) (types.FnRef, error) {
 	}
 }
 
-func (ps *MetaResolver) resolveForRuntime(targetFn string, runtime string) (types.FnRef, error) {
-	dst, ok := ps.clients[runtime]
+func (ps *MetaResolver) resolveForRuntime(ref types.FnRef) (types.FnRef, error) {
+	dst, ok := ps.clients[ref.Runtime]
 	if !ok {
 		return types.FnRef{}, ErrInvalidRuntime
 	}
-	rsv, err := dst.Resolve(targetFn)
+	rsv, err := dst.Resolve(ref)
 	if err != nil {
 		return types.FnRef{}, err
 	}
 
-	fnResolved.WithLabelValues(runtime).Inc()
+	fnResolved.WithLabelValues(ref.Runtime).Inc()
 	return types.FnRef{
-		Runtime: runtime,
-		ID:      rsv,
+		Runtime:   ref.Runtime,
+		Namespace: ref.Namespace,
+		ID:        rsv,
 	}, nil
 }
 
@@ -182,10 +177,12 @@ func resolveTask(ps Resolver, id string, task *types.TaskSpec, resolvedC chan so
 	if task == nil || resolvedC == nil {
 		return nil
 	}
+
 	t, err := ps.Resolve(task.FunctionRef)
 	if err != nil {
 		return err
 	}
+
 	resolvedC <- sourceFnRef{
 		src:   id,
 		FnRef: &t,

pkg/fnenv/resolver_test.go

@@ -47,7 +47,7 @@ func TestResolveForced(t *testing.T) {
 	resolver := NewMetaResolver(clients)
 
 	task1 := "task1"
-	task1Ref := types.NewFnRef(fooClient, "lowercase")
+	task1Ref := types.NewFnRef(fooClient, "", "lowercase")
 	task1Name := task1Ref.Format()
 	tasks := map[string]*types.TaskSpec{
 		task1: {
@@ -133,6 +133,6 @@ type MockedFunctionResolver struct {
 	Fn func(string) (string, error)
 }
 
-func (mk *MockedFunctionResolver) Resolve(fnName string) (string, error) {
-	return mk.Fn(fnName)
+func (mk *MockedFunctionResolver) Resolve(ref types.FnRef) (string, error) {
+	return mk.Fn(ref.ID)
 }

pkg/fnenv/workflows/workflows_test.go

@@ -131,7 +131,7 @@ func TestRuntime_InvokeWorkflow_Cancel(t *testing.T) {
 func TestRuntime_Invoke(t *testing.T) {
 	runtime, invocationAPI, _, cache := setup()
 
-	spec := types.NewTaskInvocationSpec("wi-123", "ti-123", types.NewFnRef("internal", "fooFn"))
+	spec := types.NewTaskInvocationSpec("wi-123", "ti-123", types.NewFnRef("internal", "", "fooFn"))
 	spec.Inputs = types.Inputs{}
 	spec.Inputs[types.InputParent] = typedvalues.MustParse("parentID")
 	output := typedvalues.MustParse("foo")

pkg/types/fnref.go

@@ -4,16 +4,24 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const (
 	RuntimeDelimiter = "://"
-	groupRuntime     = 1
-	groupRuntimeID   = 2
+)
+
+const (
+	// These are the index of capture groups for the FnRef regex
+	groupFnRef = iota
+	groupRuntime
+	groupNamespace
+	groupRuntimeID
 )
 
 var (
-	fnRefReg        = regexp.MustCompile(fmt.Sprintf("^(?:(\\w+)%s)?(\\w+)$", RuntimeDelimiter))
+	fnRefReg        = regexp.MustCompile(fmt.Sprintf("^(?:(\\w+)%s)?(?:([a-zA-Z0-9][a-zA-Z0-9_-]{1,128})/)?(\\w+)$", RuntimeDelimiter))
 	ErrInvalidFnRef = errors.New("invalid function reference")
 	ErrNoRuntime    = errors.New("function reference does not contain a runtime")
 	ErrNoRuntimeID  = errors.New("function reference does not contain a runtimeId")
@@ -25,6 +33,11 @@ func (m FnRef) Format() string {
 		runtime = m.Runtime + RuntimeDelimiter
 	}
 
+	if len(m.Namespace) > 0 {
+		return runtime + m.Namespace + `/` + m.ID
+
+	}
+
 	return runtime + m.ID
 }
 
@@ -41,13 +54,14 @@ func IsFnRef(s string) bool {
 	return err == nil
 }
 
-func NewFnRef(runtime string, id string) FnRef {
+func NewFnRef(runtime, ns, id string) FnRef {
 	if len(id) == 0 {
 		panic("function reference needs a runtime id")
 	}
 	return FnRef{
-		Runtime: runtime,
-		ID:      id,
+		Runtime:   runtime,
+		Namespace: ns,
+		ID:        id,
 	}
 }
 
@@ -57,20 +71,14 @@ func ParseFnRef(s string) (FnRef, error) {
 		return FnRef{}, ErrInvalidFnRef
 	}
 
-	if len(matches[groupRuntimeID]) == 0 {
-		return FnRef{
-			Runtime: matches[groupRuntime],
-		}, ErrNoRuntimeID
-	}
-
-	if len(matches[groupRuntime]) == 0 {
-		return FnRef{
-			ID: matches[groupRuntimeID],
-		}, ErrNoRuntime
+	ns := matches[groupNamespace]
+	if len(ns) == 0 {
+		ns = metav1.NamespaceDefault
 	}
 
 	return FnRef{
-		Runtime: matches[groupRuntime],
-		ID:      matches[groupRuntimeID],
+		Runtime:   matches[groupRuntime],
+		Namespace: ns,
+		ID:        matches[groupRuntimeID],
 	}, nil
 }

pkg/types/fnref_test.go

@@ -8,15 +8,18 @@ import (
 
 var parseCases = map[string]struct {
 	FnRef
-	err error
+	err  error
+	full string
 }{
-	"":              {FnRef{}, ErrInvalidFnRef},
-	"noRuntime":     {NewFnRef("", "noRuntime"), ErrNoRuntime},
-	"://":           {FnRef{}, ErrInvalidFnRef},
-	"://runtimeId":  {FnRef{}, ErrInvalidFnRef},
-	"runtime://":    {FnRef{}, ErrInvalidFnRef},
-	"a://b":         {NewFnRef("a", "b"), nil},
-	"http://foobar": {NewFnRef("http", "foobar"), nil},
+	"noRuntime":                         {NewFnRef("", "default", "noRuntime"), nil, "default/noRuntime"},
+	"a://b":                             {NewFnRef("a", "default", "b"), nil, "a://default/b"},
+	"http://foobar":                     {NewFnRef("http", "default", "foobar"), nil, "http://default/foobar"},
+	"fission://fission-function/foobar": {NewFnRef("fission", "fission-function", "foobar"), nil, "fission://fission-function/foobar"},
+
+	"":             {FnRef{}, ErrInvalidFnRef, ""},
+	"://":          {FnRef{}, ErrInvalidFnRef, ""},
+	"://runtimeId": {FnRef{}, ErrInvalidFnRef, ""},
+	"runtime://":   {FnRef{}, ErrInvalidFnRef, ""},
 }
 
 func TestParse(t *testing.T) {
@@ -36,7 +39,7 @@ func TestFnRef_Format(t *testing.T) {
 		t.Run(expected, func(t *testing.T) {
 			fnRef := input.FnRef
 			if !fnRef.IsEmpty() {
-				assert.Equal(t, expected, fnRef.Format())
+				assert.Equal(t, input.full, fnRef.Format())
 			}
 		})
 	}