diff --git a/controller/api_test.go b/controller/api_test.go index 7c24887e..bd68e725 100644 --- a/controller/api_test.go +++ b/controller/api_test.go @@ -24,6 +24,7 @@ import ( "net/http" "os" "reflect" + "strings" "testing" "time" @@ -69,9 +70,8 @@ func assertNotFoundFailure(err error, name string) { func assertCronSpecFails(err error) { assert(err != nil, "using an invalid cron spec must fail") - fe, ok := err.(fission.Error) - assert(ok, "error must be a fission Error") - assert(fe.Code == fission.ErrorInvalidArgument, "error must be a invalid argument error") + ok := strings.Contains(err.Error(), "not a valid cron spec") + assert(ok, "invalid cron spec must fail") } func TestFunctionApi(t *testing.T) { @@ -82,10 +82,16 @@ func TestFunctionApi(t *testing.T) { }, Spec: fission.FunctionSpec{ Environment: fission.EnvironmentReference{ - Name: "nodejs", + Name: "nodejs", + Namespace: metav1.NamespaceDefault, }, Package: fission.FunctionPackageRef{ FunctionName: "xxx", + PackageRef: fission.PackageRef{ + Namespace: metav1.NamespaceDefault, + Name: "xxx", + ResourceVersion: "12345", + }, }, }, } @@ -142,6 +148,7 @@ func TestHTTPTriggerApi(t *testing.T) { Namespace: metav1.NamespaceDefault, }, Spec: fission.HTTPTriggerSpec{ + Method: http.MethodGet, RelativeURL: "/hello", FunctionReference: fission.FunctionReference{ Type: fission.FunctionReferenceTypeFunctionName, diff --git a/controller/client/environment.go b/controller/client/environment.go index d7cc8f0c..4927f2c5 100644 --- a/controller/client/environment.go +++ b/controller/client/environment.go @@ -24,10 +24,16 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) EnvironmentCreate(env *crd.Environment) (*metav1.ObjectMeta, error) { + err := env.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Environment", err) + } + reqbody, err := json.Marshal(env) if err != nil { return nil, err @@ -78,6 +84,11 @@ func (c *Client) EnvironmentGet(m *metav1.ObjectMeta) (*crd.Environment, error) } func (c *Client) EnvironmentUpdate(env *crd.Environment) (*metav1.ObjectMeta, error) { + err := env.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Environment", err) + } + reqbody, err := json.Marshal(env) if err != nil { return nil, err diff --git a/controller/client/function.go b/controller/client/function.go index 51c68e0d..f4cc0477 100644 --- a/controller/client/function.go +++ b/controller/client/function.go @@ -24,10 +24,15 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) FunctionCreate(f *crd.Function) (*metav1.ObjectMeta, error) { + err := f.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Function", err) + } reqbody, err := json.Marshal(f) if err != nil { @@ -93,6 +98,11 @@ func (c *Client) FunctionGetRawDeployment(m *metav1.ObjectMeta) ([]byte, error) } func (c *Client) FunctionUpdate(f *crd.Function) (*metav1.ObjectMeta, error) { + err := f.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Function", err) + } + reqbody, err := json.Marshal(f) if err != nil { return nil, err diff --git a/controller/client/httptrigger.go b/controller/client/httptrigger.go index cd7cf15c..a9197f3a 100644 --- a/controller/client/httptrigger.go +++ b/controller/client/httptrigger.go @@ -24,10 +24,16 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) HTTPTriggerCreate(t *crd.HTTPTrigger) (*metav1.ObjectMeta, error) { + err := t.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("HTTPTrigger", err) + } + reqbody, err := json.Marshal(t) if err != nil { return nil, err @@ -78,6 +84,11 @@ func (c *Client) HTTPTriggerGet(m *metav1.ObjectMeta) (*crd.HTTPTrigger, error) } func (c *Client) HTTPTriggerUpdate(t *crd.HTTPTrigger) (*metav1.ObjectMeta, error) { + err := t.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("HTTPTrigger", err) + } + reqbody, err := json.Marshal(t) if err != nil { return nil, err diff --git a/controller/client/kuberneteswatchtrigger.go b/controller/client/kuberneteswatchtrigger.go index d6f97e70..aed286f0 100644 --- a/controller/client/kuberneteswatchtrigger.go +++ b/controller/client/kuberneteswatchtrigger.go @@ -29,6 +29,11 @@ import ( ) func (c *Client) WatchCreate(w *crd.KubernetesWatchTrigger) (*metav1.ObjectMeta, error) { + err := w.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("KubernetesWatchTrigger", err) + } + reqbody, err := json.Marshal(w) if err != nil { return nil, err diff --git a/controller/client/mqtrigger.go b/controller/client/mqtrigger.go index 2f937e25..19703248 100644 --- a/controller/client/mqtrigger.go +++ b/controller/client/mqtrigger.go @@ -24,10 +24,16 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) MessageQueueTriggerCreate(t *crd.MessageQueueTrigger) (*metav1.ObjectMeta, error) { + err := t.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("MessageQueueTrigger", err) + } + reqbody, err := json.Marshal(t) if err != nil { return nil, err @@ -78,6 +84,11 @@ func (c *Client) MessageQueueTriggerGet(m *metav1.ObjectMeta) (*crd.MessageQueue } func (c *Client) MessageQueueTriggerUpdate(mqTrigger *crd.MessageQueueTrigger) (*metav1.ObjectMeta, error) { + err := mqTrigger.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("MessageQueueTrigger", err) + } + reqbody, err := json.Marshal(mqTrigger) if err != nil { return nil, err diff --git a/controller/client/package.go b/controller/client/package.go index 87159aef..a6a2a487 100644 --- a/controller/client/package.go +++ b/controller/client/package.go @@ -24,10 +24,15 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) PackageCreate(f *crd.Package) (*metav1.ObjectMeta, error) { + err := f.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Package", err) + } reqbody, err := json.Marshal(f) if err != nil { @@ -79,6 +84,11 @@ func (c *Client) PackageGet(m *metav1.ObjectMeta) (*crd.Package, error) { } func (c *Client) PackageUpdate(f *crd.Package) (*metav1.ObjectMeta, error) { + err := f.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("Package", err) + } + reqbody, err := json.Marshal(f) if err != nil { return nil, err diff --git a/controller/client/timetrigger.go b/controller/client/timetrigger.go index 971d7df7..a87521fc 100644 --- a/controller/client/timetrigger.go +++ b/controller/client/timetrigger.go @@ -24,10 +24,16 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) func (c *Client) TimeTriggerCreate(t *crd.TimeTrigger) (*metav1.ObjectMeta, error) { + err := t.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("TimeTrigger", err) + } + reqbody, err := json.Marshal(t) if err != nil { return nil, err @@ -78,6 +84,11 @@ func (c *Client) TimeTriggerGet(m *metav1.ObjectMeta) (*crd.TimeTrigger, error) } func (c *Client) TimeTriggerUpdate(t *crd.TimeTrigger) (*metav1.ObjectMeta, error) { + err := t.Validate() + if err != nil { + return nil, fission.AggregateValidationErrors("TimeTrigger", err) + } + reqbody, err := json.Marshal(t) if err != nil { return nil, err diff --git a/controller/crd.go b/controller/crd.go index c8e20da3..90fbd58b 100644 --- a/controller/crd.go +++ b/controller/crd.go @@ -17,9 +17,6 @@ limitations under the License. package controller import ( - "errors" - "regexp" - "github.com/fission/fission/crd" ) @@ -30,11 +27,3 @@ func makeCRDBackedAPI() (*API, error) { } return &API{fissionClient: fissionClient, kubernetesClient: kubernetesClient}, nil } - -func validateResourceName(name string) error { - re := regexp.MustCompile(`[a-z0-9]([-a-z0-9]*[a-z0-9])?`) - if len(re.FindString(name)) != len(name) { - return errors.New("Name must consist of lower case alphanumeric characters or '-', and must start and end with an alphanumeric character (e.g. 'my-name', or '123-abc', regex used for validation is '[a-z0-9]([-a-z0-9]*[a-z0-9])?'") - } - return nil -} diff --git a/controller/environmentApi.go b/controller/environmentApi.go index 9c32e442..57cd58e7 100644 --- a/controller/environmentApi.go +++ b/controller/environmentApi.go @@ -60,12 +60,6 @@ func (a *API) EnvironmentApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(env.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - enew, err := a.fissionClient.Environments(env.Metadata.Namespace).Create(&env) if err != nil { a.respondWithError(w, err) diff --git a/controller/functionApi.go b/controller/functionApi.go index 7106a7c1..183a78f1 100644 --- a/controller/functionApi.go +++ b/controller/functionApi.go @@ -73,12 +73,6 @@ func (a *API) FunctionApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(f.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - fnew, err := a.fissionClient.Functions(f.Metadata.Namespace).Create(&f) if err != nil { a.respondWithError(w, err) diff --git a/controller/httpTriggerApi.go b/controller/httpTriggerApi.go index 111e7a1b..200e6c0e 100644 --- a/controller/httpTriggerApi.go +++ b/controller/httpTriggerApi.go @@ -74,12 +74,6 @@ func (a *API) HTTPTriggerApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(t.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - // Ensure we don't have a duplicate HTTP route defined (same URL and method) err = a.checkHTTPTriggerDuplicates(&t) if err != nil { diff --git a/controller/mqTriggerApi.go b/controller/mqTriggerApi.go index 9ecefb67..0805e76b 100644 --- a/controller/mqTriggerApi.go +++ b/controller/mqTriggerApi.go @@ -57,12 +57,6 @@ func (a *API) MessageQueueTriggerApiCreate(w http.ResponseWriter, r *http.Reques return } - err = validateResourceName(mqTrigger.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - tnew, err := a.fissionClient.MessageQueueTriggers(mqTrigger.Metadata.Namespace).Create(&mqTrigger) if err != nil { a.respondWithError(w, err) diff --git a/controller/packageApi.go b/controller/packageApi.go index 2a8c53a2..be6ed0c3 100644 --- a/controller/packageApi.go +++ b/controller/packageApi.go @@ -58,12 +58,6 @@ func (a *API) PackageApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(f.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - // Ensure size limits if len(f.Spec.Source.Literal) > 256*1024 { err := fission.MakeError(fission.ErrorInvalidArgument, "Package literal larger than 256K") diff --git a/controller/timeTriggerApi.go b/controller/timeTriggerApi.go index f1bb68c4..4ae47fa2 100644 --- a/controller/timeTriggerApi.go +++ b/controller/timeTriggerApi.go @@ -59,12 +59,6 @@ func (a *API) TimeTriggerApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(t.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - // validate _, err = cron.Parse(t.Spec.Cron) if err != nil { diff --git a/controller/watchApi.go b/controller/watchApi.go index dc96cc80..40413a6a 100644 --- a/controller/watchApi.go +++ b/controller/watchApi.go @@ -58,12 +58,6 @@ func (a *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) { return } - err = validateResourceName(watch.Metadata.Name) - if err != nil { - a.respondWithError(w, err) - return - } - // TODO check for duplicate watches wnew, err := a.fissionClient.KubernetesWatchTriggers(watch.Metadata.Namespace).Create(&watch) diff --git a/crd/types.go b/crd/types.go index 40f232ce..ebe51536 100644 --- a/crd/types.go +++ b/crd/types.go @@ -17,6 +17,7 @@ limitations under the License. package crd import ( + "github.com/hashicorp/go-multierror" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" @@ -151,14 +152,14 @@ func (ht *HTTPTrigger) GetObjectKind() schema.ObjectKind { func (w *KubernetesWatchTrigger) GetObjectKind() schema.ObjectKind { return &w.TypeMeta } -func (w *TimeTrigger) GetObjectKind() schema.ObjectKind { - return &w.TypeMeta +func (t *TimeTrigger) GetObjectKind() schema.ObjectKind { + return &t.TypeMeta } -func (w *MessageQueueTrigger) GetObjectKind() schema.ObjectKind { - return &w.TypeMeta +func (m *MessageQueueTrigger) GetObjectKind() schema.ObjectKind { + return &m.TypeMeta } -func (w *Package) GetObjectKind() schema.ObjectKind { - return &w.TypeMeta +func (p *Package) GetObjectKind() schema.ObjectKind { + return &p.TypeMeta } func (f *Function) GetObjectMeta() metav1.Object { @@ -173,14 +174,14 @@ func (ht *HTTPTrigger) GetObjectMeta() metav1.Object { func (w *KubernetesWatchTrigger) GetObjectMeta() metav1.Object { return &w.Metadata } -func (w *TimeTrigger) GetObjectMeta() metav1.Object { - return &w.Metadata +func (t *TimeTrigger) GetObjectMeta() metav1.Object { + return &t.Metadata } -func (w *MessageQueueTrigger) GetObjectMeta() metav1.Object { - return &w.Metadata +func (m *MessageQueueTrigger) GetObjectMeta() metav1.Object { + return &m.Metadata } -func (w *Package) GetObjectMeta() metav1.Object { - return &w.Metadata +func (p *Package) GetObjectMeta() metav1.Object { + return &p.Metadata } func (fl *FunctionList) GetObjectKind() schema.ObjectKind { @@ -198,11 +199,11 @@ func (wl *KubernetesWatchTriggerList) GetObjectKind() schema.ObjectKind { func (wl *TimeTriggerList) GetObjectKind() schema.ObjectKind { return &wl.TypeMeta } -func (wl *MessageQueueTriggerList) GetObjectKind() schema.ObjectKind { - return &wl.TypeMeta +func (ml *MessageQueueTriggerList) GetObjectKind() schema.ObjectKind { + return &ml.TypeMeta } -func (wl *PackageList) GetObjectKind() schema.ObjectKind { - return &wl.TypeMeta +func (pl *PackageList) GetObjectKind() schema.ObjectKind { + return &pl.TypeMeta } func (fl *FunctionList) GetListMeta() metav1.List { @@ -220,9 +221,141 @@ func (wl *KubernetesWatchTriggerList) GetListMeta() metav1.List { func (wl *TimeTriggerList) GetListMeta() metav1.List { return &wl.Metadata } -func (wl *MessageQueueTriggerList) GetListMeta() metav1.List { - return &wl.Metadata +func (ml *MessageQueueTriggerList) GetListMeta() metav1.List { + return &ml.Metadata } -func (wl *PackageList) GetListMeta() metav1.List { - return &wl.Metadata +func (pl *PackageList) GetListMeta() metav1.List { + return &pl.Metadata +} + +func validateMetadata(field string, m metav1.ObjectMeta) error { + return fission.ValidateKubeReference(field, m.Name, m.Namespace) +} + +func (p *Package) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("Package", p.Metadata), + p.Spec.Validate(), + p.Status.Validate()) + + return result.ErrorOrNil() +} + +func (pl *PackageList) Validate() error { + var result *multierror.Error + // not validate ListMeta + for _, p := range pl.Items { + result = multierror.Append(result, p.Validate()) + } + return result.ErrorOrNil() +} + +func (f *Function) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("Function", f.Metadata), + f.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (fl *FunctionList) Validate() error { + var result *multierror.Error + for _, f := range fl.Items { + result = multierror.Append(result, f.Validate()) + } + return result.ErrorOrNil() +} + +func (e *Environment) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("Environment", e.Metadata), + e.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (el *EnvironmentList) Validate() error { + var result *multierror.Error + for _, e := range el.Items { + result = multierror.Append(result, e.Validate()) + } + return result.ErrorOrNil() +} + +func (h *HTTPTrigger) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("HTTPTrigger", h.Metadata), + h.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (hl *HTTPTriggerList) Validate() error { + var result *multierror.Error + for _, h := range hl.Items { + result = multierror.Append(result, h.Validate()) + } + return result.ErrorOrNil() +} + +func (k *KubernetesWatchTrigger) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("KubernetesWatchTrigger", k.Metadata), + k.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (kl *KubernetesWatchTriggerList) Validate() error { + var result *multierror.Error + for _, k := range kl.Items { + result = multierror.Append(result, k.Validate()) + } + return result +} + +func (t *TimeTrigger) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("TimeTrigger", t.Metadata), + t.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (tl *TimeTriggerList) Validate() error { + var result *multierror.Error + for _, t := range tl.Items { + result = multierror.Append(result, t.Validate()) + } + return result.ErrorOrNil() +} + +func (m *MessageQueueTrigger) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("MessageQueueTrigger", m.Metadata), + m.Spec.Validate()) + + return result.ErrorOrNil() +} + +func (ml *MessageQueueTriggerList) Validate() error { + var result *multierror.Error + for _, m := range ml.Items { + result = multierror.Append(result, m.Validate()) + } + return result.ErrorOrNil() } diff --git a/fission/httptrigger.go b/fission/httptrigger.go index efb75354..e1ecab18 100644 --- a/fission/httptrigger.go +++ b/fission/httptrigger.go @@ -63,6 +63,7 @@ func checkFunctionExistence(fissionClient *client.Client, fnName string) { Name: fnName, Namespace: metav1.NamespaceDefault, } + _, err := fissionClient.FunctionGet(meta) if err != nil { fmt.Printf("function '%v' does not exist, use 'fission function create --name %v ...' to create the function\n", fnName, fnName) diff --git a/fission/main.go b/fission/main.go index e535ad3e..45dc01be 100644 --- a/fission/main.go +++ b/fission/main.go @@ -201,7 +201,6 @@ func main() { envBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "Build command for environment builder to build source package (optional)"} envExternalNetworkFlag := cli.BoolFlag{Name: "externalnetwork", Usage: "Allow environment access external network when istio feature enabled (optional, defaults to false)"} envTerminationGracePeriodFlag := cli.Int64Flag{Name: "graceperiod, period", Usage: "The grace time for pod to perform connection draining before termination (optional, defaults to 360 seconds)"} - envVersionFlag := cli.IntFlag{Name: "version", Usage: "Environment API version: defaults to 1 (means v1 interface)"} envSubcommands := []cli.Command{ {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag, envExternalNetworkFlag, envTerminationGracePeriodFlag, specSaveFlag}, Action: envCreate}, diff --git a/fission/migrate.go b/fission/migrate.go index 42892664..a89f845f 100644 --- a/fission/migrate.go +++ b/fission/migrate.go @@ -27,7 +27,6 @@ import ( "github.com/fission/fission" "github.com/fission/fission/controller/client" "github.com/fission/fission/crd" - "github.com/fission/fission/mqtrigger/messageQueue" ) type ( @@ -55,7 +54,7 @@ func migrateDumpTPRResource(client *client.Client, filename string) { checkErr(err, "dump watches") timeTriggers, err := client.TimeTriggerList() checkErr(err, "dump time triggers") - mqTriggers, err := client.MessageQueueTriggerList(messageQueue.NATS) + mqTriggers, err := client.MessageQueueTriggerList(fission.MessageQueueTypeNats) checkErr(err, "dump message queue triggers") tprResource := TPRResource{ diff --git a/fission/mqtrigger.go b/fission/mqtrigger.go index fc50fe45..8e3ed36a 100644 --- a/fission/mqtrigger.go +++ b/fission/mqtrigger.go @@ -27,7 +27,6 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/mqtrigger/messageQueue" ) func mqtCreate(c *cli.Context) error { @@ -42,14 +41,14 @@ func mqtCreate(c *cli.Context) error { fatal("Need a function name to create a trigger, use --function") } - mqType := c.String("mqtype") - switch mqType { + var mqType fission.MessageQueueType + switch c.String("mqtype") { case "": - mqType = messageQueue.NATS - case messageQueue.NATS: - mqType = messageQueue.NATS - case messageQueue.ASQ: - mqType = messageQueue.ASQ + mqType = fission.MessageQueueTypeNats + case fission.MessageQueueTypeNats: + mqType = fission.MessageQueueTypeNats + case fission.MessageQueueTypeASQ: + mqType = fission.MessageQueueTypeASQ default: fatal("Unknown message queue type, currently only \"nats-streaming, azure-storage-queue \" is supported") } @@ -194,9 +193,9 @@ func mqtList(c *cli.Context) error { return nil } -func checkMQTopicAvailability(mqType string, topics ...string) { +func checkMQTopicAvailability(mqType fission.MessageQueueType, topics ...string) { for _, t := range topics { - if len(t) > 0 && !messageQueue.IsTopicValid(mqType, t) { + if len(t) > 0 && !fission.IsTopicValid(mqType, t) { fatal(fmt.Sprintf("Invalid topic for %s: %s", mqType, t)) } } diff --git a/fission/upgrade.go b/fission/upgrade.go index 71f299f4..4262f2e8 100644 --- a/fission/upgrade.go +++ b/fission/upgrade.go @@ -360,7 +360,7 @@ func upgradeRestoreState(c *cli.Context) error { Metadata: *crdMetadataFromV1Metadata(&t.Metadata, v1state.NameChanges), Spec: fission.MessageQueueTriggerSpec{ FunctionReference: *functionRefFromV1Metadata(&t.Function, v1state.NameChanges), - MessageQueueType: t.MessageQueueType, + MessageQueueType: fission.MessageQueueTypeNats, // only NATS is supported at that time (v1 types) Topic: t.Topic, ResponseTopic: t.ResponseTopic, }, diff --git a/mqtrigger/messageQueue/asq_test.go b/mqtrigger/messageQueue/asq_test.go index 8eaafd44..083a9d99 100644 --- a/mqtrigger/messageQueue/asq_test.go +++ b/mqtrigger/messageQueue/asq_test.go @@ -104,7 +104,7 @@ func (m *azureHTTPClientMock) Do(req *http.Request) (*http.Response, error) { func TestNewStorageConnectionMissingAccountName(t *testing.T) { connection, err := newAzureStorageConnection(DummyRouterURL, MessageQueueConfig{ - MQType: ASQ, + MQType: fission.MessageQueueTypeASQ, Url: "", }) require.Nil(t, connection) @@ -114,7 +114,7 @@ func TestNewStorageConnectionMissingAccountName(t *testing.T) { func TestNewStorageConnectionMissingAccessKey(t *testing.T) { _ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname") connection, err := newAzureStorageConnection(DummyRouterURL, MessageQueueConfig{ - MQType: ASQ, + MQType: fission.MessageQueueTypeASQ, Url: "", }) _ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_NAME") @@ -291,7 +291,7 @@ func TestAzureStorageQueuePoisonMessage(t *testing.T) { Type: fission.FunctionReferenceTypeFunctionName, Name: FunctionName, }, - MessageQueueType: ASQ, + MessageQueueType: fission.MessageQueueTypeASQ, Topic: QueueName, ContentType: ContentType, }, @@ -434,7 +434,7 @@ func runAzureStorageQueueTest(t *testing.T, count int, output bool) { Type: fission.FunctionReferenceTypeFunctionName, Name: FunctionName, }, - MessageQueueType: ASQ, + MessageQueueType: fission.MessageQueueTypeASQ, Topic: QueueName, ResponseTopic: responseTopic, ContentType: ContentType, diff --git a/mqtrigger/messageQueue/messageQueue.go b/mqtrigger/messageQueue/messageQueue.go index b299c232..6e13e053 100644 --- a/mqtrigger/messageQueue/messageQueue.go +++ b/mqtrigger/messageQueue/messageQueue.go @@ -18,7 +18,6 @@ package messageQueue import ( "errors" - "regexp" "time" log "github.com/sirupsen/logrus" @@ -28,21 +27,12 @@ import ( "github.com/fission/fission/crd" ) -const ( - NATS string = "nats-streaming" - ASQ string = "azure-storage-queue" -) - const ( ADD_TRIGGER requestType = iota DELETE_TRIGGER GET_ALL_TRIGGERS ) -var ( - validAzureQueueName = regexp.MustCompile("^[a-z0-9][a-z0-9\\-]*[a-z0-9]$") -) - type ( messageQueueSubscription interface{} @@ -92,9 +82,9 @@ func MakeMessageQueueTriggerManager(fissionClient *crd.FissionClient, routerUrl fissionClient: fissionClient, } switch mqConfig.MQType { - case NATS: + case fission.MessageQueueTypeNats: messageQueue, err = makeNatsMessageQueue(routerUrl, mqConfig) - case ASQ: + case fission.MessageQueueTypeASQ: messageQueue, err = newAzureStorageConnection(routerUrl, mqConfig) default: err = errors.New("No matched message queue type found") @@ -231,13 +221,3 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() { time.Sleep(3 * time.Second) } } - -func IsTopicValid(mqType string, topic string) bool { - switch mqType { - case NATS: - return isTopicValidForNats(topic) - case ASQ: - return len(topic) >= 3 && len(topic) <= 63 && validAzureQueueName.MatchString(topic) - } - return false -} diff --git a/types.go b/types.go index bbd308e4..7ec6c520 100644 --- a/types.go +++ b/types.go @@ -99,6 +99,7 @@ type ( // package update, making it possible to cache the function based on its metadata. ResourceVersion string `json:"resourceversion,omitempty"` } + FunctionPackageRef struct { PackageRef PackageRef `json:"packageref"` @@ -289,11 +290,13 @@ type ( FunctionReference FunctionReference `json:"functionref"` } + MessageQueueType string + // MessageQueueTriggerSpec defines a binding from a topic in a // message queue to a function. MessageQueueTriggerSpec struct { FunctionReference FunctionReference `json:"functionref"` - MessageQueueType string `json:"messageQueueType"` + MessageQueueType MessageQueueType `json:"messageQueueType"` Topic string `json:"topic"` ResponseTopic string `json:"respTopic,omitempty"` ContentType string `json:"contentType"` @@ -387,6 +390,11 @@ const ( SharedVolumeConfigmaps = "configmaps" ) +const ( + MessageQueueTypeNats = "nats-streaming" + MessageQueueTypeASQ = "azure-storage-queue" +) + const ( // FunctionReferenceFunctionName means that the function // reference is simply by function name. diff --git a/validation.go b/validation.go new file mode 100644 index 00000000..e95fbd13 --- /dev/null +++ b/validation.go @@ -0,0 +1,458 @@ +/* +Copyright 2018 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package fission + +import ( + "fmt" + "net/http" + "regexp" + "strings" + + "github.com/hashicorp/go-multierror" + nsUtil "github.com/nats-io/nats-streaming-server/util" + "github.com/robfig/cron" + "k8s.io/apimachinery/pkg/util/validation" +) + +const ( + ErrorUnsupportedType = iota + ErrorInvalidValue + ErrorInvalidObject +) + +var ( + validAzureQueueName = regexp.MustCompile("^[a-z0-9][a-z0-9\\-]*[a-z0-9]$") +) + +type ValidationErrorType int + +// ValidationError is a custom error type for resource validation. +// It indicate which field is invalid or illegal in the fission resource. +// Also, it shows what kind of error type, bad value and detail error messages. +type ValidationError struct { + // Type of validation error. + // It indicates what kind of error of field in error output. + Type ValidationErrorType + + // Name of error field. + // Example: FunctionReference.Name + Field string + + // Error field value. + BadValue interface{} + + // Detail error message + Detail string +} + +func (e ValidationError) Error() string { + // Example error message + // Failed to create HTTP trigger: Invalid fission HTTPTrigger object: + // * FunctionReference.Name: Invalid value: findped.ts: [...] + + errMsg := fmt.Sprintf("%v: ", e.Field) + + switch e.Type { + case ErrorUnsupportedType: + errMsg += fmt.Sprintf("Unsupported type: %v", e.BadValue) + case ErrorInvalidValue: + errMsg += fmt.Sprintf("Invalid value: %v", e.BadValue) + case ErrorInvalidObject: + errMsg += fmt.Sprintf("Invalid object: %v", e.BadValue) + default: + errMsg += fmt.Sprintf("Unknown error type: %v", e.BadValue) + } + + if len(e.Detail) > 0 { + errMsg += fmt.Sprintf(": %v", e.Detail) + } + + return errMsg +} + +func AggregateValidationErrors(objName string, err error) error { + var result *multierror.Error + + result = multierror.Append(result, err) + + result.ErrorFormat = func(errs []error) string { + errMsg := fmt.Sprintf("Invalid fission %v object:\n", objName) + for _, err := range errs { + errMsg += fmt.Sprintf("* %v\n", err.Error()) + } + return errMsg + } + + return result.ErrorOrNil() +} + +func MakeValidationErr(errType ValidationErrorType, field string, val interface{}, detail ...string) ValidationError { + return ValidationError{ + Type: errType, + Field: field, + BadValue: val, + Detail: fmt.Sprintf("%v", detail), + } +} + +func ValidateKubeLabel(field string, labels map[string]string) error { + var result *multierror.Error + + for k, v := range labels { + // Example: XXX -> YYY + // KubernetesWatchTriggerSpec.LabelSelector.Key: Invalid value: XXX + // KubernetesWatchTriggerSpec.LabelSelector.Value: Invalid value: YYY + result = multierror.Append(result, + MakeValidationErr(ErrorInvalidValue, fmt.Sprintf("%v.Key", field), k, validation.IsQualifiedName(k)...), + MakeValidationErr(ErrorInvalidValue, fmt.Sprintf("%v.Value", field), v, validation.IsValidLabelValue(v)...)) + } + + return result.ErrorOrNil() +} + +func ValidateKubePort(field string, port int) error { + var result *multierror.Error + + e := validation.IsValidPortNum(port) + if len(e) > 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, field, port, e...)) + } + + return result.ErrorOrNil() +} + +func ValidateKubeName(field string, val string) error { + var result *multierror.Error + + e := validation.IsDNS1123Label(val) + if len(e) > 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, field, val, e...)) + } + + return result.ErrorOrNil() +} + +func ValidateKubeReference(refName string, name string, namespace string) error { + var result *multierror.Error + + result = multierror.Append(result, + ValidateKubeName(fmt.Sprintf("%v.Name", refName), name), + ValidateKubeName(fmt.Sprintf("%v.Namespace", refName), namespace)) + + return result.ErrorOrNil() +} + +func IsTopicValid(mqType MessageQueueType, topic string) bool { + switch mqType { + case MessageQueueTypeNats: + return nsUtil.IsChannelNameValid(topic, false) + case MessageQueueTypeASQ: + return len(topic) >= 3 && len(topic) <= 63 && validAzureQueueName.MatchString(topic) + } + return false +} + +func IsValidCronSpec(spec string) error { + _, err := cron.Parse(spec) + return err +} + +/* Resource validation function */ + +func (checksum Checksum) Validate() error { + var result *multierror.Error + + switch checksum.Type { + case ChecksumTypeSHA256: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "Checksum.Type", checksum.Type, "not a valid checksum type")) + } + + return result.ErrorOrNil() +} + +func (archive Archive) Validate() error { + var result *multierror.Error + + if len(archive.Type) > 0 { + switch archive.Type { + case ArchiveTypeLiteral, ArchiveTypeUrl: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "Archive.Type", archive.Type, "not a valid archive type")) + } + } + + if archive.Checksum != (Checksum{}) { + result = multierror.Append(result, archive.Checksum.Validate()) + } + + return result.ErrorOrNil() +} + +func (ref EnvironmentReference) Validate() error { + var result *multierror.Error + result = multierror.Append(result, ValidateKubeReference("EnvironmentReference", ref.Name, ref.Namespace)) + return result.ErrorOrNil() +} + +func (ref SecretReference) Validate() error { + var result *multierror.Error + result = multierror.Append(result, ValidateKubeReference("SecretReference", ref.Name, ref.Namespace)) + return result.ErrorOrNil() +} + +func (ref ConfigMapReference) Validate() error { + var result *multierror.Error + result = multierror.Append(result, ValidateKubeReference("ConfigMapReference", ref.Name, ref.Namespace)) + return result.ErrorOrNil() +} + +func (spec PackageSpec) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, spec.Environment.Validate()) + + for _, r := range []Archive{spec.Source, spec.Deployment} { + if len(r.URL) > 0 || len(r.Literal) > 0 { + result = multierror.Append(result, r.Validate()) + } + } + + return result.ErrorOrNil() +} + +func (sts PackageStatus) Validate() error { + var result *multierror.Error + + switch sts.BuildStatus { + case BuildStatusPending, BuildStatusRunning, BuildStatusSucceeded, BuildStatusFailed, BuildStatusNone: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "PackageStatus.BuildStatus", sts.BuildStatus, "not a valid build status")) + } + + return result.ErrorOrNil() +} + +func (ref PackageRef) Validate() error { + var result *multierror.Error + result = multierror.Append(result, ValidateKubeReference("PackageRef", ref.Name, ref.Namespace)) + return result.ErrorOrNil() +} + +func (ref FunctionPackageRef) Validate() error { + var result *multierror.Error + result = multierror.Append(result, ref.PackageRef.Validate()) + return result.ErrorOrNil() +} + +func (spec FunctionSpec) Validate() error { + var result *multierror.Error + + if spec.Environment != (EnvironmentReference{}) { + result = multierror.Append(result, spec.Environment.Validate()) + } + + if spec.Package != (FunctionPackageRef{}) { + result = multierror.Append(result, spec.Package.Validate()) + } + + for _, s := range spec.Secrets { + result = multierror.Append(result, s.Validate()) + } + for _, c := range spec.ConfigMaps { + result = multierror.Append(result, c.Validate()) + } + + if spec.InvokeStrategy != (InvokeStrategy{}) { + result = multierror.Append(result, spec.InvokeStrategy.Validate()) + } + + return result.ErrorOrNil() +} + +func (is InvokeStrategy) Validate() error { + var result *multierror.Error + + switch is.StrategyType { + case StrategyTypeExecution: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "InvokeStrategy.StrategyType", is.StrategyType, "not a valid valid strategy")) + } + + result = multierror.Append(result, is.ExecutionStrategy.Validate()) + + return result.ErrorOrNil() +} + +func (es ExecutionStrategy) Validate() error { + var result *multierror.Error + + switch es.ExecutorType { + case ExecutorTypeNewdeploy, ExecutorTypePoolmgr: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "ExecutionStrategy.ExecutorType", es.ExecutorType, "not a valid executor type")) + } + + if es.MinScale < 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "ExecutionStrategy.MinScale", es.MinScale, "minimum scale must be greater or equal to 0")) + } + + if es.MaxScale < es.MinScale { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "ExecutionStrategy.MaxScale", es.MaxScale, "maximum scale must be greater or equal to minimum scale")) + } + + if es.TargetCPUPercent <= 0 || es.TargetCPUPercent > 100 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "ExecutionStrategy.TargetCPUPercent", es.TargetCPUPercent, "TargetCPUPercent must be a value between 1 - 100")) + } + + return result.ErrorOrNil() +} + +func (ref FunctionReference) Validate() error { + var result *multierror.Error + + switch ref.Type { + case FunctionReferenceTypeFunctionName: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "FunctionReference.Type", ref.Type, "not a valid function reference type")) + } + + result = multierror.Append(result, ValidateKubeName("FunctionReference.Name", ref.Name)) + + return result.ErrorOrNil() +} + +func (runtime Runtime) Validate() error { + var result *multierror.Error + + if runtime.LoadEndpointPort > 0 { + result = multierror.Append(result, ValidateKubePort("Runtime.LoadEndpointPort", int(runtime.LoadEndpointPort))) + } + + if runtime.FunctionEndpointPort > 0 { + result = multierror.Append(result, ValidateKubePort("Runtime.FunctionEndpointPort", int(runtime.FunctionEndpointPort))) + } + + return result.ErrorOrNil() +} + +func (builder Builder) Validate() error { + // do nothing for now + return nil +} + +func (spec EnvironmentSpec) Validate() error { + var result *multierror.Error + + if spec.Version < 1 && spec.Version > 3 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "EnvironmentSpec.Version", spec.Version, "not a valid environment version")) + } + + result = multierror.Append(result, spec.Runtime.Validate()) + + if spec.Builder != (Builder{}) { + result = multierror.Append(result, spec.Builder.Validate()) + } + + if len(spec.AllowedFunctionsPerContainer) > 0 { + switch spec.AllowedFunctionsPerContainer { + case AllowedFunctionsPerContainerSingle, AllowedFunctionsPerContainerInfinite: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "EnvironmentSpec.AllowedFunctionsPerContainer", spec.AllowedFunctionsPerContainer, "not a valid value")) + } + } + + if spec.Poolsize < 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "EnvironmentSpec.Poolsize", spec.Poolsize, "Poolsize must be greater or equal to 0")) + } + + return result.ErrorOrNil() +} + +func (spec HTTPTriggerSpec) Validate() error { + var result *multierror.Error + + switch spec.Method { + case http.MethodGet, http.MethodHead, http.MethodPost, http.MethodPut, http.MethodPatch, + http.MethodDelete, http.MethodConnect, http.MethodOptions, http.MethodTrace: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "HTTPTriggerSpec.Method", spec.Method, "not a valid HTTP method")) + } + + result = multierror.Append(result, spec.FunctionReference.Validate()) + + if len(spec.Host) > 0 { + e := validation.IsDNS1123Subdomain(spec.Host) + if len(e) > 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "HTTPTriggerSpec.Host", spec.Host, e...)) + } + } + + return result.ErrorOrNil() +} + +func (spec KubernetesWatchTriggerSpec) Validate() error { + var result *multierror.Error + + switch strings.ToUpper(spec.Type) { + case "POD", "SERVICE", "REPLICATIONCONTROLLER", "JOB": + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "KubernetesWatchTriggerSpec.Type", spec.Type, "not a valid supported type")) + } + + result = multierror.Append(result, + ValidateKubeName("KubernetesWatchTriggerSpec.Namespace", spec.Namespace), + ValidateKubeLabel("KubernetesWatchTriggerSpec.LabelSelector", spec.LabelSelector), + spec.FunctionReference.Validate()) + + return result.ErrorOrNil() +} + +func (spec MessageQueueTriggerSpec) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, spec.FunctionReference.Validate()) + + switch spec.MessageQueueType { + case MessageQueueTypeNats, MessageQueueTypeASQ: // no op + default: + result = multierror.Append(result, MakeValidationErr(ErrorUnsupportedType, "MessageQueueTriggerSpec.MessageQueueType", spec.MessageQueueType, "not a supported message queue type")) + } + + if !IsTopicValid(spec.MessageQueueType, spec.Topic) { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "MessageQueueTriggerSpec.Topic", spec.Topic, "not a valid topic")) + } + + if len(spec.ResponseTopic) > 0 && !IsTopicValid(spec.MessageQueueType, spec.ResponseTopic) { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "MessageQueueTriggerSpec.ResponseTopic", spec.ResponseTopic, "not a valid topic")) + } + + return result.ErrorOrNil() +} + +func (spec TimeTriggerSpec) Validate() error { + var result *multierror.Error + + err := IsValidCronSpec(spec.Cron) + if err != nil { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "TimeTriggerSpec.Cron", spec.Cron, "not a valid cron spec")) + } + + result = multierror.Append(result, spec.FunctionReference.Validate()) + + return result.ErrorOrNil() +}