Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add validate function to crd resource and do validate before creation/update (#580)

Browse Source
This commit is contained in:
Ta-Ching Chen
2018-04-08 00:56:13 +08:00
committed by GitHub
parent 12299df811
commit d01b309fc7
26 changed files with 718 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/api_test.go
+11 -4
View File
@@ -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,
controller/client/environment.go
+11
View File
@@ -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
controller/client/function.go
+10
View File
@@ -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
controller/client/httptrigger.go
+11
View File
@@ -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
controller/client/kuberneteswatchtrigger.go
+5
View File
@@ -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
controller/client/mqtrigger.go
+11
View File
@@ -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
controller/client/package.go
+10
View File
@@ -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
controller/client/timetrigger.go
+11
View File
@@ -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
controller/crd.go
-11
View File
@@ -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
}
controller/environmentApi.go
-6
View File
@@ -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)
controller/functionApi.go
-6
View File
@@ -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)
controller/httpTriggerApi.go
-6
View File
@@ -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 {
controller/mqTriggerApi.go
-6
View File
@@ -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)
controller/packageApi.go
-6
View File
@@ -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")
controller/timeTriggerApi.go
-6
View File
@@ -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 {
controller/watchApi.go
-6
View File
@@ -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)
crd/types.go
+153 -20
View File
@@ -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()
}
fission/httptrigger.go
+1
View File
@@ -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)
fission/main.go
-1
View File
@@ -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},
fission/migrate.go
+1 -2
View File
@@ -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{
fission/mqtrigger.go
+9 -10
View File
@@ -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))
}
}
fission/upgrade.go
+1 -1
View File
@@ -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,
},
mqtrigger/messageQueue/asq_test.go
+4 -4
View File
@@ -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,
mqtrigger/messageQueue/messageQueue.go
+2 -22
View File
@@ -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
}
types.go
+9 -1
View File
@@ -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.
validation.go
+458
View File
@@ -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()
}