From 74a3a545433ebbf82fa8dabb652412f7f2e62b5c Mon Sep 17 00:00:00 2001 From: Nafisa Shazia Date: Tue, 14 Aug 2018 14:28:40 -0700 Subject: [PATCH] Recorder CRD, Records API, Redis deployment (#818) --- charts/fission-all/templates/redis.yaml | 39 ++ controller/api.go | 11 + controller/client/recorder.go | 239 +++++++++++++ controller/recorderApi.go | 148 ++++++++ controller/recordsApi.go | 95 +++++ crd/client.go | 11 +- crd/crd.go | 16 + crd/recorder.go | 121 +++++++ crd/types.go | 2 + fission/main.go | 30 ++ fission/recorder.go | 238 ++++++++++++ fission/records.go | 144 ++++++++ fission/spec.go | 4 + glide.lock | 95 +++-- glide.yaml | 4 +- pkg/apis/fission.io/v1/typefields.go | 11 + pkg/apis/fission.io/v1/types.go | 39 ++ pkg/apis/fission.io/v1/validation.go | 27 ++ .../fission.io/v1/zz_generated.deepcopy.go | 81 +++++ redis/README.md | 5 + redis/build/gen/spec.pb.go | 305 ++++++++++++++++ redis/redis.go | 123 +++++++ redis/redisApi.go | 338 ++++++++++++++++++ redis/spec.proto | 51 +++ router/functionHandler.go | 59 +++ router/functionRecorderMap.go | 63 ++++ router/httpTriggers.go | 73 +++- router/recorderController.go | 118 ++++++ router/router.go | 7 +- router/router_test.go | 6 +- router/triggerRecorderMap.go | 62 ++++ test/tests/mqtrigger/test_mqtrigger.sh | 61 ---- test/tests/recordreplay/greetings.py | 9 + test/tests/recordreplay/rendezvous.py | 7 + .../recordreplay/test_record_greetings.sh | 55 +++ test/tests/recordreplay/test_record_rv.sh | 95 +++++ .../recordreplay/test_recorder_update.sh | 100 ++++++ types.go | 1 + 38 files changed, 2770 insertions(+), 123 deletions(-) create mode 100644 charts/fission-all/templates/redis.yaml create mode 100644 controller/client/recorder.go create mode 100644 controller/recorderApi.go create mode 100644 controller/recordsApi.go create mode 100644 crd/recorder.go create mode 100644 fission/recorder.go create mode 100644 fission/records.go create mode 100644 redis/README.md create mode 100644 redis/build/gen/spec.pb.go create mode 100644 redis/redis.go create mode 100644 redis/redisApi.go create mode 100644 redis/spec.proto create mode 100644 router/functionRecorderMap.go create mode 100644 router/recorderController.go create mode 100644 router/triggerRecorderMap.go delete mode 100755 test/tests/mqtrigger/test_mqtrigger.sh create mode 100644 test/tests/recordreplay/greetings.py create mode 100644 test/tests/recordreplay/rendezvous.py create mode 100755 test/tests/recordreplay/test_record_greetings.sh create mode 100755 test/tests/recordreplay/test_record_rv.sh create mode 100755 test/tests/recordreplay/test_recorder_update.sh diff --git a/charts/fission-all/templates/redis.yaml b/charts/fission-all/templates/redis.yaml new file mode 100644 index 00000000..586d5e22 --- /dev/null +++ b/charts/fission-all/templates/redis.yaml @@ -0,0 +1,39 @@ +apiVersion: v1 +kind: Service +metadata: + name: redis + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + ports: + - port: 6379 + targetPort: 6379 + protocol: TCP + name: redis + selector: + app: redis +--- +apiVersion: apps/v1beta2 +kind: StatefulSet +metadata: + name: redis + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + selector: + matchLabels: + app: redis + serviceName: redis + replicas: 1 + template: + metadata: + labels: + app: redis + spec: + containers: + - name: redis + image: redis:3.2-alpine + imagePullPolicy: Always + ports: + - containerPort: 6379 + name: redis diff --git a/controller/api.go b/controller/api.go index 22328e9a..f25b32a2 100644 --- a/controller/api.go +++ b/controller/api.go @@ -237,6 +237,17 @@ func (api *API) Serve(port int) { r.HandleFunc("/v2/triggers/messagequeue/{mqTrigger}", api.MessageQueueTriggerApiUpdate).Methods("PUT") r.HandleFunc("/v2/triggers/messagequeue/{mqTrigger}", api.MessageQueueTriggerApiDelete).Methods("DELETE") + r.HandleFunc("/v2/recorders", api.RecorderApiList).Methods("GET") + r.HandleFunc("/v2/recorders", api.RecorderApiCreate).Methods("POST") + r.HandleFunc("/v2/recorders/{recorder}", api.RecorderApiGet).Methods("GET") + r.HandleFunc("/v2/recorders/{recorder}", api.RecorderApiUpdate).Methods("PUT") + r.HandleFunc("/v2/recorders/{recorder}", api.RecorderApiDelete).Methods("DELETE") + + r.HandleFunc("/v2/records", api.RecordsApiListAll).Methods("GET") + r.HandleFunc("/v2/records/function/{function}", api.RecordsApiFilterByFunction).Methods("GET") + r.HandleFunc("/v2/records/trigger/{trigger}", api.RecordsApiFilterByTrigger).Methods("GET") + r.HandleFunc("/v2/records/time", api.RecordsApiFilterByTime).Methods("GET") + r.HandleFunc("/v2/secrets/{secret}", api.SecretGet).Methods("GET") r.HandleFunc("/v2/configmaps/{configmap}", api.ConfigMapGet).Methods("GET") diff --git a/controller/client/recorder.go b/controller/client/recorder.go new file mode 100644 index 00000000..4861a766 --- /dev/null +++ b/controller/client/recorder.go @@ -0,0 +1,239 @@ +/* +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 client + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission/crd" + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" + "github.com/fission/fission/redis/build/gen" +) + +func (c *Client) RecorderCreate(r *crd.Recorder) (*metav1.ObjectMeta, error) { + err := r.Validate() + if err != nil { + return nil, fv1.AggregateValidationErrors("Recorder", err) + } + + reqbody, err := json.Marshal(r) + if err != nil { + return nil, err + } + + resp, err := http.Post(c.url("recorders"), "application/json", bytes.NewReader(reqbody)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleCreateResponse(resp) + if err != nil { + return nil, err + } + + var m metav1.ObjectMeta + err = json.Unmarshal(body, &m) + if err != nil { + return nil, err + } + + return &m, nil +} + +func (c *Client) RecorderGet(m *metav1.ObjectMeta) (*crd.Recorder, error) { + relativeUrl := fmt.Sprintf("recorders/%v", m.Name) + relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace) + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + var r crd.Recorder + err = json.Unmarshal(body, &r) + if err != nil { + return nil, err + } + + return &r, nil +} + +func (c *Client) RecorderUpdate(recorder *crd.Recorder) (*metav1.ObjectMeta, error) { + err := recorder.Validate() + if err != nil { + return nil, fv1.AggregateValidationErrors("Recorder", err) + } + + reqbody, err := json.Marshal(recorder) + if err != nil { + return nil, err + } + relativeUrl := fmt.Sprintf("recorders/%v", recorder.Metadata.Name) + + resp, err := c.put(relativeUrl, "application/json", reqbody) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + var m metav1.ObjectMeta + err = json.Unmarshal(body, &m) + if err != nil { + return nil, err + } + return &m, nil +} + +func (c *Client) RecorderDelete(m *metav1.ObjectMeta) error { + relativeUrl := fmt.Sprintf("recorders/%v", m.Name) + relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace) + return c.delete(relativeUrl) +} + +func (c *Client) RecorderList(ns string) ([]crd.Recorder, error) { + relativeUrl := "recorders" + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + recorders := make([]crd.Recorder, 0) + err = json.Unmarshal(body, &recorders) + if err != nil { + return nil, err + } + + return recorders, nil +} + +// TODO: Move to different file? +func (c *Client) RecordsByFunction(function string) ([]*redisCache.RecordedEntry, error) { + relativeUrl := fmt.Sprintf("records/function/%v", function) + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + records := make([]*redisCache.RecordedEntry, 0) + err = json.Unmarshal(body, &records) + if err != nil { + return nil, err + } + + return records, nil +} + +func (c *Client) RecordsAll() ([]*redisCache.RecordedEntry, error) { + relativeUrl := "records" + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + records := make([]*redisCache.RecordedEntry, 0) + err = json.Unmarshal(body, &records) + if err != nil { + return nil, err + } + + return records, nil +} + +func (c *Client) RecordsByTrigger(trigger string) ([]*redisCache.RecordedEntry, error) { + relativeUrl := fmt.Sprintf("records/trigger/%v", trigger) + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + records := make([]*redisCache.RecordedEntry, 0) + err = json.Unmarshal(body, &records) + if err != nil { + return nil, err + } + + return records, nil +} + +func (c *Client) RecordsByTime(from string, to string) ([]*redisCache.RecordedEntry, error) { + relativeUrl := "records/time" + relativeUrl += fmt.Sprintf("?from=%v&to=%v", from, to) + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + records := make([]*redisCache.RecordedEntry, 0) + err = json.Unmarshal(body, &records) + if err != nil { + return nil, err + } + + return records, nil +} diff --git a/controller/recorderApi.go b/controller/recorderApi.go new file mode 100644 index 00000000..357c821b --- /dev/null +++ b/controller/recorderApi.go @@ -0,0 +1,148 @@ +/* +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 controller + +import ( + "encoding/json" + "io/ioutil" + "net/http" + + "github.com/gorilla/mux" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission" + "github.com/fission/fission/crd" +) + +func (a *API) RecorderApiList(w http.ResponseWriter, r *http.Request) { + recorders, err := a.fissionClient.Recorders(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + resp, err := json.Marshal(recorders.Items) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} + +func (a *API) RecorderApiCreate(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + + if err != nil { + a.respondWithError(w, err) + return + } + + var recorder crd.Recorder + err = json.Unmarshal(body, &recorder) + if err != nil { + a.respondWithError(w, err) + return + } + + tnew, err := a.fissionClient.Recorders(recorder.Metadata.Namespace).Create(&recorder) + if err != nil { + a.respondWithError(w, err) + return + } + + resp, err := json.Marshal(tnew.Metadata) + if err != nil { + a.respondWithError(w, err) + return + } + w.WriteHeader(http.StatusCreated) + a.respondWithSuccess(w, resp) +} + +func (a *API) RecorderApiGet(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + name := vars["recorder"] + ns := a.extractQueryParamFromRequest(r, "namespace") + if len(ns) == 0 { + ns = metav1.NamespaceDefault + } + + recorder, err := a.fissionClient.Recorders(ns).Get(name) + if err != nil { + a.respondWithError(w, err) + return + } + resp, err := json.Marshal(recorder) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} + +func (a *API) RecorderApiUpdate(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + name := vars["recorder"] + + body, err := ioutil.ReadAll(r.Body) + if err != nil { + a.respondWithError(w, err) + return + } + + var recorder crd.Recorder + err = json.Unmarshal(body, &recorder) + if err != nil { + a.respondWithError(w, err) + return + } + + if name != recorder.Metadata.Name { + err = fission.MakeError(fission.ErrorInvalidArgument, "Recorder name doesn't match URL") + a.respondWithError(w, err) + return + } + + rnew, err := a.fissionClient.Recorders(recorder.Metadata.Namespace).Update(&recorder) + if err != nil { + a.respondWithError(w, err) + return + } + + resp, err := json.Marshal(rnew.Metadata) + if err != nil { + a.respondWithError(w, err) + return + } + + a.respondWithSuccess(w, resp) +} + +func (a *API) RecorderApiDelete(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + name := vars["recorder"] + ns := a.extractQueryParamFromRequest(r, "namespace") + if len(ns) == 0 { + ns = metav1.NamespaceDefault + } + + err := a.fissionClient.Recorders(ns).Delete(name, &metav1.DeleteOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, []byte("")) +} diff --git a/controller/recordsApi.go b/controller/recordsApi.go new file mode 100644 index 00000000..40a6aff6 --- /dev/null +++ b/controller/recordsApi.go @@ -0,0 +1,95 @@ +/* +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 controller + +import ( + "net/http" + + "github.com/gorilla/mux" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission/redis" +) + +func (a *API) RecordsApiListAll(w http.ResponseWriter, r *http.Request) { + resp, err := redis.RecordsListAll() + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} + +func (a *API) RecordsApiFilterByFunction(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + query := vars["function"] + + recorders, err := a.fissionClient.Recorders(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + + triggers, err := a.fissionClient.HTTPTriggers(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + + resp, err := redis.RecordsFilterByFunction(query, recorders, triggers) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} + +func (a *API) RecordsApiFilterByTrigger(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + query := vars["trigger"] + + recorders, err := a.fissionClient.Recorders(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + + triggers, err := a.fissionClient.HTTPTriggers(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + a.respondWithError(w, err) + return + } + + resp, err := redis.RecordsFilterByTrigger(query, recorders, triggers) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} + +func (a *API) RecordsApiFilterByTime(w http.ResponseWriter, r *http.Request) { + from := r.FormValue("from") + to := r.FormValue("to") + + resp, err := redis.RecordsFilterByTime(from, to) + if err != nil { + a.respondWithError(w, err) + return + } + a.respondWithSuccess(w, resp) +} diff --git a/crd/client.go b/crd/client.go index 57389f42..42ded767 100644 --- a/crd/client.go +++ b/crd/client.go @@ -153,6 +153,13 @@ func configureClient(config *rest.Config) { &metav1.ListOptions{}, &metav1.DeleteOptions{}, ) + scheme.AddKnownTypes( + groupversion, + &Recorder{}, + &RecorderList{}, + &metav1.ListOptions{}, + &metav1.DeleteOptions{}, + ) return nil }) schemeBuilder.AddToScheme(scheme.Scheme) @@ -208,10 +215,12 @@ func (fc *FissionClient) TimeTriggers(ns string) TimeTriggerInterface { func (fc *FissionClient) MessageQueueTriggers(ns string) MessageQueueTriggerInterface { return MakeMessageQueueTriggerInterface(fc.crdClient, ns) } +func (fc *FissionClient) Recorders(ns string) RecorderInterface { + return MakeRecorderInterface(fc.crdClient, ns) +} func (fc *FissionClient) Packages(ns string) PackageInterface { return MakePackageInterface(fc.crdClient, ns) } - func (fc *FissionClient) WaitForCRDs() error { return waitForCRDs(fc.crdClient) } diff --git a/crd/crd.go b/crd/crd.go index 8dd2acb1..cf4c5bc2 100644 --- a/crd/crd.go +++ b/crd/crd.go @@ -160,6 +160,22 @@ func EnsureFissionCRDs(clientset *apiextensionsclient.Clientset) error { }, }, }, + // Recorders + { + ObjectMeta: metav1.ObjectMeta{ + Name: "recorders.fission.io", + }, + Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ + Group: crdGroupName, + Version: crdVersion, + Scope: apiextensionsv1beta1.NamespaceScoped, + Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ + Kind: "Recorder", + Plural: "recorders", + Singular: "recorder", + }, + }, + }, // Packages: archives containing source or binaries for one or more functions { ObjectMeta: metav1.ObjectMeta{ diff --git a/crd/recorder.go b/crd/recorder.go new file mode 100644 index 00000000..7e2a1405 --- /dev/null +++ b/crd/recorder.go @@ -0,0 +1,121 @@ +/* +Copyright 2016 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 crd + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" +) + +type ( + RecorderInterface interface { + Create(*Recorder) (*Recorder, error) + Get(name string) (*Recorder, error) + Update(*Recorder) (*Recorder, error) + Delete(name string, opts *metav1.DeleteOptions) error + List(opts metav1.ListOptions) (*RecorderList, error) + Watch(opts metav1.ListOptions) (watch.Interface, error) + } + + recorderClient struct { + client *rest.RESTClient + namespace string + } +) + +func MakeRecorderInterface(crdClient *rest.RESTClient, namespace string) RecorderInterface { + return &recorderClient{ + client: crdClient, + namespace: namespace, + } +} + +func (rc *recorderClient) Create(r *Recorder) (*Recorder, error) { + var result Recorder + err := rc.client.Post(). + Resource("recorders"). + Namespace("default"). + Body(r). + Do().Into(&result) + if err != nil { + return nil, err + } + return &result, nil +} + +func (rc *recorderClient) Get(name string) (*Recorder, error) { + var result Recorder + err := rc.client.Get(). + Resource("recorders"). + Namespace(rc.namespace). + Name(name). + Do().Into(&result) + if err != nil { + return nil, err + } + return &result, nil +} + +func (rc *recorderClient) Update(r *Recorder) (*Recorder, error) { + var result Recorder + err := rc.client.Put(). + Resource("recorders"). + Namespace(rc.namespace). + Name(r.Metadata.Name). + Body(r). + Do().Into(&result) + if err != nil { + return nil, err + } + return &result, nil +} + +func (rc *recorderClient) Delete(name string, opts *metav1.DeleteOptions) error { + return rc.client.Delete(). + Namespace(rc.namespace). + Resource("recorders"). + Name(name). + Body(opts). + Do(). + Error() +} + +func (rc *recorderClient) List(opts metav1.ListOptions) (*RecorderList, error) { + var result RecorderList + err := rc.client.Get(). + Namespace(rc.namespace). + Resource("recorders"). + VersionedParams(&opts, scheme.ParameterCodec). + Do(). + Into(&result) + if err != nil { + return nil, err + } + return &result, nil +} + +func (rc *recorderClient) Watch(opts metav1.ListOptions) (watch.Interface, error) { + return rc.client.Get(). + Prefix("watch"). + Namespace(rc.namespace). + Resource("recorders"). + VersionedParams(&opts, scheme.ParameterCodec). + Watch() +} diff --git a/crd/types.go b/crd/types.go index 7ac9b16d..6cda534d 100644 --- a/crd/types.go +++ b/crd/types.go @@ -35,4 +35,6 @@ type ( TimeTriggerList = fv1.TimeTriggerList MessageQueueTrigger = fv1.MessageQueueTrigger MessageQueueTriggerList = fv1.MessageQueueTriggerList + Recorder = fv1.Recorder + RecorderList = fv1.RecorderList ) diff --git a/fission/main.go b/fission/main.go index 02bc40bf..938096e1 100644 --- a/fission/main.go +++ b/fission/main.go @@ -103,6 +103,7 @@ func main() { envNamespaceFlag := cli.StringFlag{Name: "envNamespace, envns", Value: metav1.NamespaceDefault, Usage: "Namespace for environment object"} pkgNamespaceFlag := cli.StringFlag{Name: "pkgNamespace, pkgns", Value: metav1.NamespaceDefault, Usage: "Namespace for package object"} triggerNamespaceFlag := cli.StringFlag{Name: "triggerNamespace, triggerns", Value: metav1.NamespaceDefault, Usage: "Namespace for trigger object"} + recorderNamespaceFlag := cli.StringFlag{Name: "recorderNamespace, recorderns", Value: metav1.NamespaceDefault, Usage: "Namespace for recorder object"} // trigger method and url flags (used in function and route CLIs) htMethodFlag := cli.StringFlag{Name: "method", Value: "GET", Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD"} @@ -197,6 +198,33 @@ func main() { {Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag, triggerNamespaceFlag}, Action: mqtList}, } + // Recorders + recNameFlag := cli.StringFlag{Name: "name", Usage: "Recorder name"} + recFnFlag := cli.StringFlag{Name: "function", Usage: "Record Function name(s): --function=fnA"} + recTriggersFlag := cli.StringSliceFlag{Name: "trigger", Usage: "Record Trigger name(s): --trigger=trigger1,trigger2,trigger3"} + //recRetentionPolFlag := cli.StringFlag{Name: "retention", Usage: "Retention policy (number of days)"} + //recEvictionPolFlag := cli.StringFlag{Name: "eviction", Usage: "Eviction policy (default LRU)"} + recEnabled := cli.BoolFlag{Name: "enable", Usage: "Enable recorder"} + recDisabled := cli.BoolFlag{Name: "disable", Usage: "Disable recorder"} + recSubcommands := []cli.Command{ + {Name: "create", Aliases: []string{"add"}, Usage: "Create recorder", Flags: []cli.Flag{recNameFlag, recFnFlag, recTriggersFlag, specSaveFlag}, Action: recorderCreate}, + {Name: "get", Usage: "Get recorder", Flags: []cli.Flag{recNameFlag}, Action: recorderGet}, + {Name: "update", Usage: "Update recorder", Flags: []cli.Flag{recNameFlag, recFnFlag, recTriggersFlag, recEnabled, recDisabled}, Action: recorderUpdate}, + {Name: "delete", Usage: "Delete recorder", Flags: []cli.Flag{recNameFlag, recorderNamespaceFlag}, Action: recorderDelete}, + {Name: "list", Usage: "List recorders", Flags: []cli.Flag{}, Action: recorderList}, + } + + // View records + filterTimeFrom := cli.StringFlag{Name: "from", Usage: "Filter records by time interval; specify start of interval"} + filterTimeTo := cli.StringFlag{Name: "to", Usage: "Filter records by time interval; specify end of interval"} + filterFunction := cli.StringFlag{Name: "function", Usage: "Filter records by function"} + filterTrigger := cli.StringFlag{Name: "trigger", Usage: "Filter records by trigger"} + verbosityFlag := cli.BoolFlag{Name: "v", Usage: "Toggle verbosity -- view more detailed requests/responses"} + vvFlag := cli.BoolFlag{Name: "vv", Usage: "Toggle verbosity -- view raw requests/responses"} + recViewSubcommands := []cli.Command{ + {Name: "view", Usage: "View existing records", Flags: []cli.Flag{filterTimeTo, filterTimeFrom, filterFunction, filterTrigger, verbosityFlag, vvFlag}, Action: recordsView}, + } + // environments envNameFlag := cli.StringFlag{Name: "name", Usage: "Environment name"} envPoolsizeFlag := cli.IntFlag{Name: "poolsize", Value: 3, Usage: "Size of the pool"} @@ -275,6 +303,8 @@ func main() { {Name: "httptrigger", Aliases: []string{"ht", "route"}, Usage: "Manage HTTP triggers (routes) for functions", Subcommands: htSubcommands}, {Name: "timetrigger", Aliases: []string{"tt", "timer"}, Usage: "Manage Time triggers (timers) for functions", Subcommands: ttSubcommands}, {Name: "mqtrigger", Aliases: []string{"mqt", "messagequeue"}, Usage: "Manage message queue triggers for functions", Subcommands: mqtSubcommands}, + {Name: "recorder", Usage: "Manage recorders for functions", Subcommands: recSubcommands, Hidden: true}, + {Name: "records", Usage: "View records with optional filters", Subcommands: recViewSubcommands, Hidden: true}, {Name: "environment", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands}, {Name: "watch", Aliases: []string{"w"}, Usage: "Manage watches", Subcommands: wSubCommands}, {Name: "package", Aliases: []string{"pkg"}, Usage: "Manage packages", Subcommands: pkgSubCommands}, diff --git a/fission/recorder.go b/fission/recorder.go new file mode 100644 index 00000000..be5d3f01 --- /dev/null +++ b/fission/recorder.go @@ -0,0 +1,238 @@ +/* +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 + + tttp://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 main + +import ( + "fmt" + "log" + "os" + "strings" + "text/tabwriter" + + "github.com/satori/go.uuid" + "github.com/urfave/cli" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission" + "github.com/fission/fission/crd" +) + +func recorderCreate(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + recName := c.String("name") + if len(recName) == 0 { + recName = uuid.NewV4().String() + } + fnName := c.String("function") + triggersOriginal := c.StringSlice("trigger") + + // Function XOR triggers can be given + if len(fnName) == 0 && len(triggersOriginal) == 0 { + log.Fatal("Need to specify at least one function or one trigger, use --function, --trigger") + } + if len(fnName) != 0 && len(triggersOriginal) != 0 { + log.Fatal("Can specify either one function or one or more triggers, but not both") + } + + // TODO: Validate here or elsewhere that all triggers belong to the same namespace + + var triggers []string + if len(triggersOriginal) != 0 { + ts := strings.Split(triggersOriginal[0], ",") + for _, name := range ts { + triggers = append(triggers, name) + } + } + // TODO: Define appropriate set of policies and defaults + //retPolicy := c.String("retention") + //evictPolicy := c.String("eviction") + + recorder := &crd.Recorder{ + Metadata: metav1.ObjectMeta{ + Name: recName, + Namespace: "default", + }, + Spec: fission.RecorderSpec{ + Name: recName, + Function: fnName, + Triggers: triggers, + RetentionPolicy: "Permanent", // TODO: Implement customizable policies for expiration of records + EvictionPolicy: "None", + Enabled: true, + }, + } + + // If we're writing a spec, don't call the API + if c.Bool("spec") { + specFile := fmt.Sprintf("recorder-%v.yaml", recName) + err := specSave(*recorder, specFile) + checkErr(err, "create recorder spec") + return nil + } + + _, err := client.RecorderCreate(recorder) + checkErr(err, "create recorder") + + fmt.Printf("recorder '%s' created\n", recName) + return err +} + +func recorderGet(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + recName := c.String("name") + + recorder, err := client.RecorderGet(&metav1.ObjectMeta{ + Name: recName, + Namespace: "default", + }) + + checkErr(err, "get recorder") + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", + "NAME", "ENABLED", "FUNCTION", "TRIGGERS", "RETENTION_POLICY", "EVICTION_POLICY") + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", + recorder.Metadata.Name, recorder.Spec.Enabled, recorder.Spec.Function, recorder.Spec.Triggers, recorder.Spec.RetentionPolicy, recorder.Spec.EvictionPolicy) + w.Flush() + + return nil +} + +func recorderUpdate(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + recName := c.String("name") + enable := c.Bool("enable") + disable := c.Bool("disable") + //retPolicy := c.String("retention") + //evictPolicy := c.String("eviction") + triggers := c.StringSlice("trigger") + function := c.String("function") + + if enable && disable { + log.Fatal("Cannot enable and disable a recorder simultaneously.") + } + + // Prevent enable or disable while trying to update other fields. These flags must be standalone. + if enable || disable { + if len(triggers) > 0 || len(function) > 0 { + log.Fatal("Enabling or disabling a recorder with other (non-name) flags set is not supported.") + } + } else if len(triggers) == 0 && len(function) == 0 { + log.Fatal("Need to specify either a function or trigger(s) for this recorder") + } + + if len(recName) == 0 { + log.Fatal("Need name of recorder, use --name") + } + + recorder, err := client.RecorderGet(&metav1.ObjectMeta{ + Name: recName, + Namespace: "default", + }) + + updated := false + + // TODO: Additional validation on type of supported retention policy, eviction policy + + //if len(retPolicy) > 0 { + // recorder.Spec.RetentionPolicy = retPolicy + // updated = true + //} + //if len(evictPolicy) > 0 { + // recorder.Spec.EvictionPolicy = evictPolicy + // updated = true + //} + if enable { + recorder.Spec.Enabled = true + updated = true + } + + if disable { + recorder.Spec.Enabled = false + updated = true + } + + if len(triggers) > 0 { + var newTriggers []string + triggs := strings.Split(triggers[0], ",") + for _, name := range triggs { + newTriggers = append(newTriggers, name) + } + recorder.Spec.Triggers = newTriggers + updated = true + } + + if len(function) > 0 { + recorder.Spec.Function = function + updated = true + } + + if !updated { + log.Fatal("Nothing to update. Use --function, --triggers, --enable or --disable") + } + + _, err = client.RecorderUpdate(recorder) + checkErr(err, "update recorder") + + fmt.Printf("recorder '%v' updated\n", recName) + return nil +} + +func recorderDelete(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + recName := c.String("name") + + if len(recName) == 0 { + log.Fatal("Need name of recorder to delete, use --name") + } + + recNs := c.String("recorderns") + + err := client.RecorderDelete(&metav1.ObjectMeta{ + Name: recName, + Namespace: recNs, + }) + + checkErr(err, "delete recorder") + + fmt.Printf("recorder '%v' deleted\n", recName) + return nil +} + +func recorderList(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + recorders, err := client.RecorderList("default") + checkErr(err, "list recorders") + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", + "NAME", "ENABLED", "FUNCTIONS", "TRIGGERS", "RETENTION_POLICY", "EVICTION_POLICY") + for _, r := range recorders { + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", + r.Metadata.Name, r.Spec.Enabled, r.Spec.Function, r.Spec.Triggers, r.Spec.RetentionPolicy, r.Spec.EvictionPolicy) + } + w.Flush() + + return nil +} diff --git a/fission/records.go b/fission/records.go new file mode 100644 index 00000000..2114b908 --- /dev/null +++ b/fission/records.go @@ -0,0 +1,144 @@ +/* +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 + + tttp://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 main + +import ( + "fmt" + "log" + "os" + "text/tabwriter" + + "github.com/urfave/cli" + + "github.com/fission/fission/redis/build/gen" +) + +func recordsView(c *cli.Context) error { + var verbosity int + if c.Bool("v") && c.Bool("vv") { + log.Fatal("Conflicting verbosity levels, use either --v or --vv") + } + if c.Bool("v") { + verbosity = 1 + } + if c.Bool("vv") { + verbosity = 2 + } + + function := c.String("function") + trigger := c.String("trigger") + from := c.String("from") + to := c.String("to") + + //Refuse multiple filters for now + if multipleFiltersSpecified(function, trigger, from+to) { + log.Fatal("Maximum of one filter is currently supported, either --function, --trigger, or --from,--to") + } + + if len(function) != 0 { + return recordsByFunction(function, verbosity, c) + } + if len(trigger) != 0 { + return recordsByTrigger(trigger, verbosity, c) + } + if len(from) != 0 && len(to) != 0 { + return recordsByTime(from, to, verbosity, c) + } + err := recordsAll(verbosity, c) + checkErr(err, "view records") + return nil +} + +func recordsAll(verbosity int, c *cli.Context) error { + fc := getClient(c.GlobalString("server")) + + records, err := fc.RecordsAll() + checkErr(err, "view records") + + showRecords(records, verbosity) + + return nil +} + +func recordsByTrigger(trigger string, verbosity int, c *cli.Context) error { + fc := getClient(c.GlobalString("server")) + + records, err := fc.RecordsByTrigger(trigger) + checkErr(err, "view records") + + showRecords(records, verbosity) + + return nil +} + +// TODO: More accurate function name (function filter) +func recordsByFunction(function string, verbosity int, c *cli.Context) error { + fc := getClient(c.GlobalString("server")) + + records, err := fc.RecordsByFunction(function) + checkErr(err, "view records") + + showRecords(records, verbosity) + + return nil +} + +func recordsByTime(from string, to string, verbosity int, c *cli.Context) error { + fc := getClient(c.GlobalString("server")) + + records, err := fc.RecordsByTime(from, to) + checkErr(err, "view records") + + showRecords(records, verbosity) + + return nil +} + +func showRecords(records []*redisCache.RecordedEntry, verbosity int) { + w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + + if verbosity == 1 { + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n", + "REQUID", "REQUEST METHOD", "FUNCTION", "RESPONSE STATUS", "TRIGGER") + for _, record := range records { + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n", + record.ReqUID, record.Req.Method, record.Req.Header["X-Fission-Function-Name"], record.Resp.Status, record.Trigger) + } + } else if verbosity == 2 { + for _, record := range records { + fmt.Println(record) + } + } else { + fmt.Fprintf(w, "%v\n", + "REQUID") + for _, record := range records { + fmt.Fprintf(w, "%v\n", + record.ReqUID) + } + } + w.Flush() +} + +func multipleFiltersSpecified(entries ...string) bool { + var specified int + for _, entry := range entries { + if len(entry) > 0 { + specified += 1 + } + } + return specified > 1 +} diff --git a/fission/spec.go b/fission/spec.go index 8225c4da..aed26f9f 100644 --- a/fission/spec.go +++ b/fission/spec.go @@ -1723,6 +1723,10 @@ func specSave(resource interface{}, specFile string) error { typedres.TypeMeta.APIVersion = SPEC_API_VERSION typedres.TypeMeta.Kind = "TimeTrigger" data, err = yaml.Marshal(typedres) + case crd.Recorder: + typedres.TypeMeta.APIVersion = SPEC_API_VERSION + typedres.TypeMeta.Kind = "Recorder" + data, err = yaml.Marshal(typedres) default: return fmt.Errorf("can't save resource %#v", resource) } diff --git a/glide.lock b/glide.lock index 142d9d03..d33ddf37 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 1f595f5cdf8dc629d9046f82869cb77811dac21ce5f58e4f81e89d21181f6c34 -updated: 2018-05-24T22:39:34.860206+08:00 +hash: b26f975bf145378e939db833fa4b6a142fa04b884463d8173622b30f95191d42 +updated: 2018-08-13T13:12:57.58715-07:00 imports: - name: cloud.google.com/go version: 3b1ae45394a234c385be014e9a488f2bb6eef821 @@ -22,7 +22,7 @@ imports: subpackages: - quantile - name: github.com/coreos/etcd - version: 6a265731e10a5137b991c1aa3a83ecefdd149d50 + version: f87b566248bb0713a56dc55bc545aa5aad17ace0 subpackages: - client - name: github.com/davecgh/go-spew @@ -47,8 +47,10 @@ imports: - internal - internal/errors - internal/prefix +- name: github.com/dustin/go-humanize + version: 9f541cc9db5d55bce703bd99987c9d5cb8eea45e - name: github.com/fsnotify/fsnotify - version: 4da3e2cfbabc9f751898f250b49f2439785783a1 + version: c2828203cd70a50dcccfb2761f8b1f8ceef9a8e9 - name: github.com/ghodss/yaml version: 73d445a93680fa1a78ae23a5839bad48f32ba1ee - name: github.com/gogo/protobuf @@ -58,15 +60,10 @@ imports: - proto - protoc-gen-gogo/descriptor - sortkeys -- name: github.com/golang/freetype - version: e2365dfdc4a05e4b8299a783240d4a7d5a65d4e4 - subpackages: - - raster - - truetype - name: github.com/golang/glog version: 44145f04b68cf362d9c4df2182967c2275eaefed - name: github.com/golang/protobuf - version: 1643683e1b54a9e88ad26d98f81400c8c9d9f4f9 + version: b4deda0973fb4c70b50d226b1af49f3da59f5265 subpackages: - proto - ptypes @@ -74,7 +71,12 @@ imports: - ptypes/duration - ptypes/timestamp - name: github.com/golang/snappy - version: 553a641470496b2327abcac10b36396bd98e45c9 + version: 2e65f85255dbc3072edf28d6b5b8efc472979f5a +- name: github.com/gomodule/redigo + version: 2cd21d9966bf7ff9ae091419744f0b3fb0fecace + subpackages: + - internal + - redis - name: github.com/google/gofuzz version: 44d81051d367757e1c7c6a5a86423ece9afcf63c - name: github.com/googleapis/gnostic @@ -95,17 +97,17 @@ imports: - name: github.com/gorilla/context version: 08b5f424b9271eedf6f9f0ce86cb9396ed337a42 - name: github.com/gorilla/handlers - version: 90663712d74cb411cbef281bc1e08c19d1a76145 + version: 7e0847f9db758cdebd26c149d0ae9d5d0b9c98ce - name: github.com/gorilla/mux version: e3702bed27f0d39777b0b37b664b6280e8ef8fbf - name: github.com/graymeta/stow - version: abb68c488872b06c5453865fa59f4818b4ea13a4 + version: 7b5498c561bbfeb17e413eb96c9edaf88af3855f subpackages: - local - name: github.com/hashicorp/errwrap - version: 7554cd9344cec97297fa6649b055a8c98c2a1e55 + version: d6c0cd88035724dd42e0f335ae30161c20575ecc - name: github.com/hashicorp/go-multierror - version: b7773ae218740a7be65057fc60b366a49b538a44 + version: 3d5d8f294aa03d8e98859feac328afbdf1ae0703 - name: github.com/hashicorp/golang-lru version: a0d98a5f288019575c6d1f4bb1573fef2d1fcdc4 subpackages: @@ -113,7 +115,7 @@ imports: - name: github.com/howeyc/gopass version: bf9dde6d0d2c004a008c27aaee91170c786f6db8 - name: github.com/imdario/mergo - version: 9d5f1277e9a8ed20c3684bda8fde67c05628518c + version: 33882c6bfe701aca0ff1472aa8b4ebd6135a560d - name: github.com/influxdata/influxdb version: b7bb7e8359642b6e071735b50ae41f5eb343fd42 subpackages: @@ -121,7 +123,7 @@ imports: - models - pkg/escape - name: github.com/json-iterator/go - version: 13f86432b882000a51c6e610c620974462691a97 + version: f2b4162afba35581b6d4a50d3b8f34e33c144682 - name: github.com/marstr/guid version: 8bdf7d1a087ccc975cf37dd6507da50698fd19ca - name: github.com/matttproud/golang_protobuf_extensions @@ -129,33 +131,35 @@ imports: subpackages: - pbutil - name: github.com/mholt/archiver - version: 26cf5bb32d07aa4e8d0de15f56ce516f4641d7df + version: e4ef56d48eb029648b0e895bb0b6a393ef0829c3 +- name: github.com/modern-go/concurrent + version: bacd9c7ef1dd9b15be4a9909b8ac7a4e313eec94 +- name: github.com/modern-go/reflect2 + version: 05fbef0ca5da472bbf96c9322b84a53edc03c9fd - name: github.com/nats-io/go-nats version: 2485387d6ede89c1c8c3445cd1935fd41a2e9ee9 subpackages: - encoders/builtin - util - name: github.com/nats-io/go-nats-streaming - version: 6e620057a207bd61e992c1c5b6a2de7b6a4cb010 + version: e15a53f85e4932540600a16b56f6c4f65f58176f subpackages: - pb - name: github.com/nats-io/nats-streaming-server - version: 6026da1b7c444bf9f1d1b1d3ed14e435aabf02bc + version: 63e2c334b66dba3edade0047625a25c0d8f18f80 subpackages: - spb - util - name: github.com/nats-io/nuid - version: 289cccf02c178dc782430d534e3c1f5b72af807f + version: 3024a71c3cbe30667286099921591e6fcc328230 - name: github.com/nwaples/rardecode version: e06696f847aeda6f39a8f0b7cdff193b7690aef6 - name: github.com/pierrec/lz4 - version: ed8d4cc3b461464e69798080a0092bd028910298 -- name: github.com/pierrec/xxHash - version: a0006b13c722f7f12368c00a3d3c2ae8a999a0c6 + version: 6b9367c9ff401dbc54fabce3fb8d972e799b702d subpackages: - - xxHash32 + - internal/xxh32 - name: github.com/pkg/errors - version: f15c970de5b76fac0b59abb32d62c17cc7bed265 + version: 816c9085562cd7ee03e7f8188a1cfd942858cded - name: github.com/prometheus/client_golang version: c5b7fccd204277076155f10851dad72b76a49317 subpackages: @@ -176,46 +180,31 @@ imports: subpackages: - xfs - name: github.com/robfig/cron - version: 2315d5715e36303a941d907f038da7f7c44c773b + version: b41be1df696709bb6395fe435af20370037c0b4c - name: github.com/satori/go.uuid version: f58768cc1a7a7e77a3bd49e98cdd21419399b6a3 - name: github.com/sirupsen/logrus version: 68cec9f21fbf3ea8d8f98c044bc6ce05f17b267a - name: github.com/spf13/pflag - version: 4c012f6dcd9546820e378d0bdda4d8fc772cdfea + version: 583c0c0531f06d5278b7d917446061adc344b5cd - name: github.com/stretchr/testify - version: 12b6f73e6084dad08a7c6e575284b177ecafbc71 + version: f35b8ab0b5a2cef36673838d662e249dd9c94686 subpackages: - assert - mock - require - name: github.com/ulikunitz/xz - version: 0c6b41e72360850ca4f98dc341fd999726ea007f + version: 636d36a76670e6c700f22fd5f4588679ff2896c4 subpackages: - internal/hash - internal/xlog - lzma - name: github.com/urfave/cli version: cfb38830724cc34fedffe9a2a29fb54fa9169cd1 -- name: github.com/wcharczuk/go-chart - version: 9e3a080aa3e7573281cf8d65a55305e1148d857d - subpackages: - - drawing - - matrix - - roboto - - seq - - util - name: golang.org/x/crypto version: 81e90905daefcd6fd217b62423c0908922eadb30 subpackages: - ssh/terminal -- name: golang.org/x/image - version: f315e440302883054d0c2bd85486878cb4f8572c - subpackages: - - draw - - font - - math/f64 - - math/fixed - name: golang.org/x/net version: 1c05540f6879653db88113bc4a2b70aec4bd491f subpackages: @@ -249,7 +238,7 @@ imports: subpackages: - rate - name: golang.org/x/tools - version: 1937f90a1bb43667aff4059b1bab13eb15121e8e + version: 87c7dcbd5db6be1a938380ce9944ed2299806701 subpackages: - imports - name: google.golang.org/appengine @@ -271,7 +260,7 @@ imports: - name: gopkg.in/yaml.v2 version: 670d4cfef0544295bc27a114dbac37980d83185a - name: k8s.io/api - version: 590a9173e3b65d74e907fcfd94b78465cf314760 + version: 0f11257a8a25954878633ebdc9841c67d8f83bdb subpackages: - admissionregistration/v1alpha1 - admissionregistration/v1beta1 @@ -302,7 +291,7 @@ imports: - storage/v1alpha1 - storage/v1beta1 - name: k8s.io/apiextensions-apiserver - version: 7fbced0db3c3378efac13578ef9d75512c582b03 + version: f584b16eb23bd2a3fd292a027d698d95db427c5d subpackages: - pkg/apis/apiextensions - pkg/apis/apiextensions/v1beta1 @@ -310,7 +299,7 @@ imports: - pkg/client/clientset/clientset/scheme - pkg/client/clientset/clientset/typed/apiextensions/v1beta1 - name: k8s.io/apimachinery - version: 31dade610c053669d8054bfd847da657251e8c1a + version: e386b2658ed20923da8cc9250e552f082899a1ee subpackages: - pkg/api/errors - pkg/api/meta @@ -420,15 +409,13 @@ imports: - util/jsonpath - util/retry - name: k8s.io/code-generator - version: d9b16e114e8c31761e26efcdee0a42cf066fbf58 + version: caff7734d7c07b6211025f0898aa6168e5bc874d - name: k8s.io/gengo - version: 01a732e01d00cb9a81bb0ca050d3e6d2b947927b + version: c42f3cdacc394f43077ff17e327d1b351c0304e4 testImports: - name: github.com/pmezard/go-difflib version: d8ed2627bdf02c080bf22230dbb337003b7aba2d subpackages: - difflib - name: github.com/stretchr/objx - version: 8a3f7159479fbc75b30357fbc48f380b7320f08e -- name: github.com/dustin/go-humanize - version: 02af3965c54e8cacf948b97fef38925c4120652c + version: b8b73a35e9830ae509858c10dec5866b4d5c8bff diff --git a/glide.yaml b/glide.yaml index 42c4adaf..045f572c 100644 --- a/glide.yaml +++ b/glide.yaml @@ -62,9 +62,11 @@ import: - package: github.com/davecgh/go-spew version: ~1.1.0 - package: github.com/imdario/mergo - version: ~0.3.2 + version: v0.3.3 - package: github.com/hashicorp/go-multierror - package: github.com/hashicorp/errwrap - package: github.com/prometheus/client_golang version: v0.8.0 - package: github.com/dustin/go-humanize +- package: github.com/golang/protobuf/proto + version: v1.1.0 diff --git a/pkg/apis/fission.io/v1/typefields.go b/pkg/apis/fission.io/v1/typefields.go index 879157cf..88773033 100644 --- a/pkg/apis/fission.io/v1/typefields.go +++ b/pkg/apis/fission.io/v1/typefields.go @@ -311,6 +311,17 @@ type ( ContentType string `json:"contentType"` } + // RecorderSpec defines a policy for recording requests and responses + // to a function, that can be later inspected or replayed. + RecorderSpec struct { + Name string `json:"name"` + Function string `json:"function"` + Triggers []string `json:"triggers"` + RetentionPolicy string `json:"retentionPolicy"` + EvictionPolicy string `json:"evictionPolicy"` + Enabled bool `json:"enabled"` + } + // TimeTrigger invokes the specific function at a time or // times specified by a cron string. TimeTriggerSpec struct { diff --git a/pkg/apis/fission.io/v1/types.go b/pkg/apis/fission.io/v1/types.go index 40e0b9f0..0ce1fd9a 100644 --- a/pkg/apis/fission.io/v1/types.go +++ b/pkg/apis/fission.io/v1/types.go @@ -152,6 +152,21 @@ type ( Items []MessageQueueTrigger `json:"items"` } + + // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + Recorder struct { + metav1.TypeMeta `json:",inline"` + Metadata metav1.ObjectMeta `json:"metadata"` + Spec RecorderSpec `json:"spec"` + } + + // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + RecorderList struct { + metav1.TypeMeta `json:",inline"` + Metadata metav1.ListMeta `json:"metadata"` + + Items []Recorder `json:"items"` + } ) // Each CRD type needs: @@ -185,6 +200,10 @@ func (p *Package) GetObjectKind() schema.ObjectKind { return &p.TypeMeta } +func (r *Recorder) GetObjectKind() schema.ObjectKind { + return &r.TypeMeta +} + func (f *Function) GetObjectMeta() metav1.Object { return &f.Metadata } @@ -207,6 +226,10 @@ func (p *Package) GetObjectMeta() metav1.Object { return &p.Metadata } +func (r *Recorder) GetObjectMeta() metav1.Object { + return &r.Metadata +} + func (fl *FunctionList) GetObjectKind() schema.ObjectKind { return &fl.TypeMeta } @@ -228,6 +251,9 @@ func (ml *MessageQueueTriggerList) GetObjectKind() schema.ObjectKind { func (pl *PackageList) GetObjectKind() schema.ObjectKind { return &pl.TypeMeta } +func (rl *RecorderList) GetObjectKind() schema.ObjectKind { + return &rl.TypeMeta +} func (fl *FunctionList) GetListMeta() metav1.ListInterface { return &fl.Metadata @@ -250,6 +276,9 @@ func (ml *MessageQueueTriggerList) GetListMeta() metav1.ListInterface { func (pl *PackageList) GetListMeta() metav1.ListInterface { return &pl.Metadata } +func (rl *RecorderList) GetListMeta() metav1.ListInterface { + return &rl.Metadata +} func validateMetadata(field string, m metav1.ObjectMeta) error { return ValidateKubeReference(field, m.Name, m.Namespace) @@ -382,3 +411,13 @@ func (ml *MessageQueueTriggerList) Validate() error { } return result.ErrorOrNil() } + +func (r *Recorder) Validate() error { + var result *multierror.Error + + result = multierror.Append(result, + validateMetadata("Recorder", r.Metadata), + r.Spec.Validate()) + + return result.ErrorOrNil() +} diff --git a/pkg/apis/fission.io/v1/validation.go b/pkg/apis/fission.io/v1/validation.go index 18e440a5..b3cd4623 100644 --- a/pkg/apis/fission.io/v1/validation.go +++ b/pkg/apis/fission.io/v1/validation.go @@ -446,6 +446,33 @@ func (spec MessageQueueTriggerSpec) Validate() error { return result.ErrorOrNil() } +func (spec RecorderSpec) Validate() error { + var result *multierror.Error + + // TODO: Function validation + //if len(spec.Function.Name) != 0 { + // result = multierror.Append(result, spec.Function.Validate()) + //} + + // TODO: Triggers validation + //for _, trigger := range spec.Triggers { + // result = multierror.Append(result, trigger.Validate()) + //} + + if len(spec.Name) == 0 { + result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "RecorderSpec.Name", spec.Name, "not a valid name")) + } + //if len(spec.RetentionPolicy) == 0 { + // result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "RecorderSpec.RetentionPolicy", spec.Name, "not a valid retention policy")) + //} + //if len(spec.EvictionPolicy) == 0 { + // result = multierror.Append(result, MakeValidationErr(ErrorInvalidValue, "RecorderSpec.EvictionPolicy", spec.Name, "not a valid eviction policy")) + //} + + //log.Info("This is the RecorderSpec validation result: %v", result) + return result.ErrorOrNil() +} + func (spec TimeTriggerSpec) Validate() error { var result *multierror.Error diff --git a/pkg/apis/fission.io/v1/zz_generated.deepcopy.go b/pkg/apis/fission.io/v1/zz_generated.deepcopy.go index fb92db82..4756dbf3 100644 --- a/pkg/apis/fission.io/v1/zz_generated.deepcopy.go +++ b/pkg/apis/fission.io/v1/zz_generated.deepcopy.go @@ -705,6 +705,87 @@ func (in *PackageStatus) DeepCopy() *PackageStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *Recorder) DeepCopyInto(out *Recorder) { + *out = *in + out.TypeMeta = in.TypeMeta + in.Metadata.DeepCopyInto(&out.Metadata) + in.Spec.DeepCopyInto(&out.Spec) + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Recorder. +func (in *Recorder) DeepCopy() *Recorder { + if in == nil { + return nil + } + out := new(Recorder) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *Recorder) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RecorderList) DeepCopyInto(out *RecorderList) { + *out = *in + out.TypeMeta = in.TypeMeta + out.Metadata = in.Metadata + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]Recorder, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecorderList. +func (in *RecorderList) DeepCopy() *RecorderList { + if in == nil { + return nil + } + out := new(RecorderList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *RecorderList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RecorderSpec) DeepCopyInto(out *RecorderSpec) { + *out = *in + if in.Triggers != nil { + in, out := &in.Triggers, &out.Triggers + *out = make([]string, len(*in)) + copy(*out, *in) + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecorderSpec. +func (in *RecorderSpec) DeepCopy() *RecorderSpec { + if in == nil { + return nil + } + out := new(RecorderSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Runtime) DeepCopyInto(out *Runtime) { *out = *in diff --git a/redis/README.md b/redis/README.md new file mode 100644 index 00000000..81dd1cd2 --- /dev/null +++ b/redis/README.md @@ -0,0 +1,5 @@ +To compile protocol buffer definitions, you need to have installed the standard C++ implementation of protocol buffers and the Go compiler plugin, protoc-gen-go. +See [here](https://github.com/golang/protobuf) for installation instructions. + +Run the following command within this directory: +`protoc --go_out=build/gen *.proto` diff --git a/redis/build/gen/spec.pb.go b/redis/build/gen/spec.pb.go new file mode 100644 index 00000000..7dcd46f5 --- /dev/null +++ b/redis/build/gen/spec.pb.go @@ -0,0 +1,305 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// source: spec.proto + +package redisCache + +import proto "github.com/golang/protobuf/proto" +import fmt "fmt" +import math "math" + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package + +type Request struct { + Method string `protobuf:"bytes,1,opt,name=Method,proto3" json:"Method,omitempty"` + URL map[string]string `protobuf:"bytes,2,rep,name=URL,proto3" json:"URL,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + Header map[string]string `protobuf:"bytes,3,rep,name=Header,proto3" json:"Header,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + Host string `protobuf:"bytes,4,opt,name=Host,proto3" json:"Host,omitempty"` + Form map[string]string `protobuf:"bytes,5,rep,name=Form,proto3" json:"Form,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + PostForm map[string]string `protobuf:"bytes,6,rep,name=PostForm,proto3" json:"PostForm,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Request) Reset() { *m = Request{} } +func (m *Request) String() string { return proto.CompactTextString(m) } +func (*Request) ProtoMessage() {} +func (*Request) Descriptor() ([]byte, []int) { + return fileDescriptor_spec_ce9436f0047fa204, []int{0} +} +func (m *Request) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Request.Unmarshal(m, b) +} +func (m *Request) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Request.Marshal(b, m, deterministic) +} +func (dst *Request) XXX_Merge(src proto.Message) { + xxx_messageInfo_Request.Merge(dst, src) +} +func (m *Request) XXX_Size() int { + return xxx_messageInfo_Request.Size(m) +} +func (m *Request) XXX_DiscardUnknown() { + xxx_messageInfo_Request.DiscardUnknown(m) +} + +var xxx_messageInfo_Request proto.InternalMessageInfo + +func (m *Request) GetMethod() string { + if m != nil { + return m.Method + } + return "" +} + +func (m *Request) GetURL() map[string]string { + if m != nil { + return m.URL + } + return nil +} + +func (m *Request) GetHeader() map[string]string { + if m != nil { + return m.Header + } + return nil +} + +func (m *Request) GetHost() string { + if m != nil { + return m.Host + } + return "" +} + +func (m *Request) GetForm() map[string]string { + if m != nil { + return m.Form + } + return nil +} + +func (m *Request) GetPostForm() map[string]string { + if m != nil { + return m.PostForm + } + return nil +} + +type Response struct { + Status string `protobuf:"bytes,7,opt,name=Status,proto3" json:"Status,omitempty"` + StatusCode int32 `protobuf:"varint,8,opt,name=StatusCode,proto3" json:"StatusCode,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Response) Reset() { *m = Response{} } +func (m *Response) String() string { return proto.CompactTextString(m) } +func (*Response) ProtoMessage() {} +func (*Response) Descriptor() ([]byte, []int) { + return fileDescriptor_spec_ce9436f0047fa204, []int{1} +} +func (m *Response) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Response.Unmarshal(m, b) +} +func (m *Response) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Response.Marshal(b, m, deterministic) +} +func (dst *Response) XXX_Merge(src proto.Message) { + xxx_messageInfo_Response.Merge(dst, src) +} +func (m *Response) XXX_Size() int { + return xxx_messageInfo_Response.Size(m) +} +func (m *Response) XXX_DiscardUnknown() { + xxx_messageInfo_Response.DiscardUnknown(m) +} + +var xxx_messageInfo_Response proto.InternalMessageInfo + +func (m *Response) GetStatus() string { + if m != nil { + return m.Status + } + return "" +} + +func (m *Response) GetStatusCode() int32 { + if m != nil { + return m.StatusCode + } + return 0 +} + +type UniqueRequest struct { + Req *Request `protobuf:"bytes,10,opt,name=Req,proto3" json:"Req,omitempty"` + Resp *Response `protobuf:"bytes,11,opt,name=Resp,proto3" json:"Resp,omitempty"` + Trigger string `protobuf:"bytes,12,opt,name=Trigger,proto3" json:"Trigger,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *UniqueRequest) Reset() { *m = UniqueRequest{} } +func (m *UniqueRequest) String() string { return proto.CompactTextString(m) } +func (*UniqueRequest) ProtoMessage() {} +func (*UniqueRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_spec_ce9436f0047fa204, []int{2} +} +func (m *UniqueRequest) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_UniqueRequest.Unmarshal(m, b) +} +func (m *UniqueRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_UniqueRequest.Marshal(b, m, deterministic) +} +func (dst *UniqueRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_UniqueRequest.Merge(dst, src) +} +func (m *UniqueRequest) XXX_Size() int { + return xxx_messageInfo_UniqueRequest.Size(m) +} +func (m *UniqueRequest) XXX_DiscardUnknown() { + xxx_messageInfo_UniqueRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_UniqueRequest proto.InternalMessageInfo + +func (m *UniqueRequest) GetReq() *Request { + if m != nil { + return m.Req + } + return nil +} + +func (m *UniqueRequest) GetResp() *Response { + if m != nil { + return m.Resp + } + return nil +} + +func (m *UniqueRequest) GetTrigger() string { + if m != nil { + return m.Trigger + } + return "" +} + +// We need this because we don't store the ReqUID in the above Message (UniqueRequest) +// and we don't store the UID there because the key in Redis is the ReqUID -- including it in the value would be +// unnecessary duplication +// but we need it here because we want to display to the user the ReqResponses for a given ReqUID +type RecordedEntry struct { + ReqUID string `protobuf:"bytes,13,opt,name=ReqUID,proto3" json:"ReqUID,omitempty"` + Req *Request `protobuf:"bytes,14,opt,name=Req,proto3" json:"Req,omitempty"` + Resp *Response `protobuf:"bytes,15,opt,name=Resp,proto3" json:"Resp,omitempty"` + Trigger string `protobuf:"bytes,16,opt,name=Trigger,proto3" json:"Trigger,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *RecordedEntry) Reset() { *m = RecordedEntry{} } +func (m *RecordedEntry) String() string { return proto.CompactTextString(m) } +func (*RecordedEntry) ProtoMessage() {} +func (*RecordedEntry) Descriptor() ([]byte, []int) { + return fileDescriptor_spec_ce9436f0047fa204, []int{3} +} +func (m *RecordedEntry) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_RecordedEntry.Unmarshal(m, b) +} +func (m *RecordedEntry) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_RecordedEntry.Marshal(b, m, deterministic) +} +func (dst *RecordedEntry) XXX_Merge(src proto.Message) { + xxx_messageInfo_RecordedEntry.Merge(dst, src) +} +func (m *RecordedEntry) XXX_Size() int { + return xxx_messageInfo_RecordedEntry.Size(m) +} +func (m *RecordedEntry) XXX_DiscardUnknown() { + xxx_messageInfo_RecordedEntry.DiscardUnknown(m) +} + +var xxx_messageInfo_RecordedEntry proto.InternalMessageInfo + +func (m *RecordedEntry) GetReqUID() string { + if m != nil { + return m.ReqUID + } + return "" +} + +func (m *RecordedEntry) GetReq() *Request { + if m != nil { + return m.Req + } + return nil +} + +func (m *RecordedEntry) GetResp() *Response { + if m != nil { + return m.Resp + } + return nil +} + +func (m *RecordedEntry) GetTrigger() string { + if m != nil { + return m.Trigger + } + return "" +} + +func init() { + proto.RegisterType((*Request)(nil), "redisCache.Request") + proto.RegisterMapType((map[string]string)(nil), "redisCache.Request.FormEntry") + proto.RegisterMapType((map[string]string)(nil), "redisCache.Request.HeaderEntry") + proto.RegisterMapType((map[string]string)(nil), "redisCache.Request.PostFormEntry") + proto.RegisterMapType((map[string]string)(nil), "redisCache.Request.URLEntry") + proto.RegisterType((*Response)(nil), "redisCache.Response") + proto.RegisterType((*UniqueRequest)(nil), "redisCache.UniqueRequest") + proto.RegisterType((*RecordedEntry)(nil), "redisCache.RecordedEntry") +} + +func init() { proto.RegisterFile("spec.proto", fileDescriptor_spec_ce9436f0047fa204) } + +var fileDescriptor_spec_ce9436f0047fa204 = []byte{ + // 392 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x93, 0x5f, 0xab, 0xd3, 0x30, + 0x18, 0xc6, 0xe9, 0xda, 0x75, 0xdd, 0x5b, 0xab, 0x23, 0x8e, 0x11, 0x86, 0x7f, 0x6a, 0x41, 0xe8, + 0x55, 0xc1, 0x09, 0xce, 0x3f, 0x78, 0xe3, 0x54, 0x26, 0x4c, 0x90, 0x68, 0x3f, 0x40, 0x5d, 0x5f, + 0xb6, 0xa1, 0x2e, 0x6b, 0x92, 0x0a, 0x3b, 0x1f, 0xe3, 0x7c, 0xdb, 0x73, 0x77, 0x68, 0xd2, 0xee, + 0x6c, 0xd0, 0x8b, 0xd3, 0xbb, 0xf7, 0x4d, 0x9e, 0xdf, 0x9b, 0xe7, 0x49, 0x5a, 0x00, 0x79, 0xc0, + 0x75, 0x72, 0x10, 0x5c, 0x71, 0x02, 0x02, 0xf3, 0x9d, 0x5c, 0x64, 0xeb, 0x2d, 0x46, 0x37, 0x36, + 0x0c, 0x18, 0x16, 0x25, 0x4a, 0x45, 0x26, 0xe0, 0x7e, 0x47, 0xb5, 0xe5, 0x39, 0xb5, 0x42, 0x2b, + 0x1e, 0xb2, 0xba, 0x23, 0x09, 0xd8, 0x29, 0x5b, 0xd1, 0x5e, 0x68, 0xc7, 0xfe, 0xec, 0x49, 0x72, + 0x47, 0x27, 0x35, 0x99, 0xa4, 0x6c, 0xf5, 0x65, 0xaf, 0xc4, 0x91, 0x55, 0x42, 0x32, 0x07, 0x77, + 0x89, 0x59, 0x8e, 0x82, 0xda, 0x1a, 0x79, 0xde, 0x86, 0x18, 0x85, 0xa1, 0x6a, 0x39, 0x21, 0xe0, + 0x2c, 0xb9, 0x54, 0xd4, 0xd1, 0xc7, 0xeb, 0x9a, 0xbc, 0x02, 0xe7, 0x2b, 0x17, 0xff, 0x68, 0x5f, + 0x8f, 0x7a, 0xda, 0x36, 0xaa, 0xda, 0x37, 0x83, 0xb4, 0x94, 0x7c, 0x04, 0xef, 0x07, 0x97, 0x4a, + 0x63, 0xae, 0xc6, 0x5e, 0xb4, 0x61, 0x8d, 0xc6, 0xa0, 0x27, 0x64, 0xfa, 0x06, 0xbc, 0x26, 0x0f, + 0x19, 0x81, 0xfd, 0x07, 0x8f, 0xf5, 0x7d, 0x54, 0x25, 0x19, 0x43, 0xff, 0x7f, 0xf6, 0xb7, 0x44, + 0xda, 0xd3, 0x6b, 0xa6, 0x79, 0xdf, 0x7b, 0x6b, 0x4d, 0xdf, 0x81, 0x7f, 0x16, 0xaa, 0x13, 0x3a, + 0x87, 0xe1, 0xc9, 0x49, 0x27, 0xf0, 0x03, 0x04, 0x17, 0x31, 0xba, 0xc0, 0xd1, 0x27, 0xf0, 0x18, + 0xca, 0x03, 0xdf, 0x4b, 0xac, 0xde, 0xfe, 0xa7, 0xca, 0x54, 0x29, 0xe9, 0xc0, 0xbc, 0xbd, 0xe9, + 0xc8, 0x33, 0x00, 0x53, 0x2d, 0x78, 0x8e, 0xd4, 0x0b, 0xad, 0xb8, 0xcf, 0xce, 0x56, 0xa2, 0x2b, + 0x08, 0xd2, 0xfd, 0xae, 0x28, 0xb1, 0xf9, 0x88, 0x5e, 0x82, 0xcd, 0xb0, 0xa0, 0x10, 0x5a, 0xb1, + 0x3f, 0x7b, 0xdc, 0x72, 0xef, 0xac, 0xda, 0x27, 0x31, 0x38, 0xd5, 0xd9, 0xd4, 0xd7, 0xba, 0xf1, + 0xa5, 0xce, 0x78, 0x62, 0x5a, 0x41, 0x28, 0x0c, 0x7e, 0x89, 0xdd, 0x66, 0x83, 0x82, 0x3e, 0xd0, + 0xd6, 0x9a, 0x36, 0xba, 0xb6, 0x20, 0x60, 0xb8, 0xe6, 0x22, 0xc7, 0xdc, 0xa4, 0x9f, 0x80, 0xcb, + 0xb0, 0x48, 0xbf, 0x7d, 0xa6, 0x81, 0x49, 0x61, 0xba, 0xc6, 0xd4, 0xc3, 0x7b, 0x9a, 0x7a, 0xd4, + 0xc5, 0xd4, 0xe8, 0xc2, 0xd4, 0x6f, 0x57, 0xff, 0x63, 0xaf, 0x6f, 0x03, 0x00, 0x00, 0xff, 0xff, + 0x0f, 0xb0, 0x49, 0xaf, 0x71, 0x03, 0x00, 0x00, +} diff --git a/redis/redis.go b/redis/redis.go new file mode 100644 index 00000000..79ed5722 --- /dev/null +++ b/redis/redis.go @@ -0,0 +1,123 @@ +/* +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 redis + +import ( + "encoding/json" + "fmt" + "net/http" + "net/url" + "os" + "strings" + + "github.com/golang/protobuf/proto" + "github.com/gomodule/redigo/redis" + log "github.com/sirupsen/logrus" + + "github.com/fission/fission/redis/build/gen" +) + +func NewClient() redis.Conn { + redisIP := os.Getenv("REDIS_SERVICE_HOST") // TODO: Do this here or somewhere earlier? + redisPort := os.Getenv("REDIS_SERVICE_PORT") + redisUrl := fmt.Sprintf("%s:%s", redisIP, redisPort) + + if len(redisUrl) == 0 { + log.Error("Could not reach Redis in cluster at IP ", redisUrl) + return nil + } + + c, err := redis.Dial("tcp", redisUrl) + if err != nil { + log.Error("Could not connect to Redis: %v\n", err) + return nil + } + return c +} + +func Record(triggerName string, recorderName string, reqUID string, request *http.Request, originalUrl url.URL, payload string, response *http.Response, namespace string, timestamp int64) { + // Case where the function should not have been recorded + if len(reqUID) == 0 { + return + } + + fullPath := originalUrl.String() + escPayload := string(json.RawMessage(payload)) + + client := NewClient() + if client == nil { + return + } + + url := make(map[string]string) + url["Host"] = request.URL.Host + url["Path"] = fullPath + url["Payload"] = escPayload + + header := make(map[string]string) + for key, value := range request.Header { + header[key] = strings.Join(value, ",") + } + + form := make(map[string]string) + for key, value := range request.Form { + form[key] = strings.Join(value, ",") + } + + postForm := make(map[string]string) + for key, value := range request.PostForm { + postForm[key] = strings.Join(value, ",") + } + + req := &redisCache.Request{ + Method: request.Method, + URL: url, + Header: header, + Host: request.Host, // Proxied host? + Form: form, + PostForm: postForm, + } + + resp := &redisCache.Response{ + Status: response.Status, + StatusCode: int32(response.StatusCode), + } + + ureq := &redisCache.UniqueRequest{ + Req: req, + Resp: resp, + Trigger: triggerName, + } + + data, err := proto.Marshal(ureq) + if err != nil { + log.Error("Error marshalling request: ", err) + return + } + + _, err = client.Do("HMSET", reqUID, "ReqResponse", data, "Timestamp", timestamp, "Trigger", triggerName) + if err != nil { + log.Error("Error saving request: ", err) + return + } + + _, err = client.Do("LPUSH", recorderName, reqUID) + if err != nil { + log.Error("Error saving recorder-request pair: ", err) + return + } +} diff --git a/redis/redisApi.go b/redis/redisApi.go new file mode 100644 index 00000000..7df08425 --- /dev/null +++ b/redis/redisApi.go @@ -0,0 +1,338 @@ +/* +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 redis + +import ( + "encoding/json" + "errors" + "strconv" + "strings" + "time" + + "github.com/golang/protobuf/proto" + "github.com/gomodule/redigo/redis" + log "github.com/sirupsen/logrus" + + "github.com/fission/fission/crd" + "github.com/fission/fission/redis/build/gen" +) + +func RecordsListAll() ([]byte, error) { + client := NewClient() + if client == nil { + return []byte{}, errors.New("failed to create redis client") + } + + iter := 0 + var filtered []*redisCache.RecordedEntry + + for { + // Each scan yields only a subset of all keys which is why we keep an iter. When iter == 0, + // Redis tells us there are no keys left to traverse. + arr, err := redis.Values(client.Do("SCAN", iter)) + if err != nil { + return []byte{}, err + } + // SCAN return value is an array of two values: the first value is the new cursor to use in the next call, + // the second value is an array of elements. + iter, _ = redis.Int(arr[0], nil) + keys, _ := redis.Strings(arr[1], nil) + for _, key := range keys { + if strings.HasPrefix(key, "REQ") { + val, err := redis.Bytes(client.Do("HGET", key, "ReqResponse")) + if err != nil { + log.Error("Error retrieving request from Redis: ", err) + return []byte{}, err + } + entry, err := deserializeReqResponse(val, key) + if err != nil { + log.Error("Error deserializing request: ", err) + return []byte{}, err + } + filtered = append(filtered, entry) + } + } + if iter == 0 { + break + } + } + + resp, err := json.Marshal(filtered) + if err != nil { + return []byte{}, err + } + return resp, nil +} + +// Input: `from` (hours ago, between 0 [today] and 5) and `to` (same units) +// Note: Fractional values don't seem to work -- document that for the user +func RecordsFilterByTime(from string, to string) ([]byte, error) { + rangeStart, rangeEnd, err := obtainInterval(from, to) + log.Debug("Interval inferred: ", rangeStart, rangeEnd) + + if rangeStart >= rangeEnd { + log.Error("Invalid chronology") + return []byte{}, err + } + + client := NewClient() + if client == nil { + return []byte{}, errors.New("failed to create redis client") + } + + iter := 0 + var filtered []*redisCache.RecordedEntry + + for { + arr, err := redis.Values(client.Do("SCAN", iter)) + if err != nil { + return []byte{}, err + } + // SCAN return value is an array of two values: the first value is the new cursor to use in the next call, + // the second value is an array of elements. + iter, _ = redis.Int(arr[0], nil) + keys, _ := redis.Strings(arr[1], nil) + for _, key := range keys { + if strings.HasPrefix(key, "REQ") { + val, err := redis.Strings(client.Do("HMGET", key, "Timestamp")) + if err != nil { + log.Error("Error retrieving timestamp from Redis: ", err) + return []byte{}, err + } + tsO, err := strconv.Atoi(val[0]) + if err != nil { + log.Error("Error converting timestamp to int: ", err) + return []byte{}, err + } + ts := int64(tsO) + if ts >= rangeStart && ts <= rangeEnd { + val2, err := redis.Bytes(client.Do("HGET", key, "ReqResponse")) + if err != nil { + log.Error("Error retrieving request from Redis: ", err) + return []byte{}, err + } + entry, err := deserializeReqResponse(val2, key) + if err != nil { + log.Error("Error deserializing request: ", err) + return []byte{}, err + } + filtered = append(filtered, entry) + } + } + } + + if iter == 0 { + break + } + } + + resp, err := json.Marshal(filtered) + if err != nil { + return []byte{}, err + } + return resp, nil +} + +func RecordsFilterByTrigger(queriedTriggerName string, recorders *crd.RecorderList, triggers *crd.HTTPTriggerList) ([]byte, error) { + matchingRecorders := make(map[string]bool) + + // Implicit triggers: + // Sometimes triggers are not explicitly attached to recorders but we still want to be able to + // filter records by those triggers; we do so by identifying the functionReference the queriedTriggerName trigger has + // and finding recorder(s) for that function + + var correspFunction string + for _, trigger := range triggers.Items { + if trigger.Metadata.Name == queriedTriggerName { + correspFunction = trigger.Spec.FunctionReference.Name + break + } + } + + for _, recorder := range recorders.Items { + if len(recorder.Spec.Triggers) > 0 { + if includesTrigger(recorder.Spec.Triggers, queriedTriggerName) { + matchingRecorders[recorder.Spec.Name] = true + } + } + if recorder.Spec.Function == correspFunction { + matchingRecorders[recorder.Spec.Name] = true + } + } + + client := NewClient() + if client == nil { + return []byte{}, errors.New("failed to create redis client") + } + + var filtered []*redisCache.RecordedEntry + + // TODO: Account for old/not-yet-deleted entries in the recorder lists + for key := range matchingRecorders { + val, err := redis.Strings(client.Do("LRANGE", key, "0", "-1")) // TODO: Prefix that distinguishes recorder lists + if err != nil { + // TODO: Handle deleted recorder? Or is this a non-issue because our list of recorders is up to date? + return []byte{}, err + } + for _, reqUID := range val { + val, err := redis.Strings(client.Do("HMGET", reqUID, "Trigger")) // 1-to-1 reqUID - trigger? + if err != nil { + log.Error("Error retrieving trigger for a request from Redis: ", err) + return []byte{}, err + } + if val[0] == queriedTriggerName { + // TODO: Reconsider multiple commands + val, err := redis.Bytes(client.Do("HGET", reqUID, "ReqResponse")) + if err != nil { + log.Error("Error retrieving request from Redis: ", err) + return []byte{}, err + } + entry, err := deserializeReqResponse(val, reqUID) + if err != nil { + log.Error("Error deserializing request: ", err) + return []byte{}, err + } + filtered = append(filtered, entry) + } + } + } + + resp, err := json.Marshal(filtered) + if err != nil { + return []byte{}, err + } + return resp, nil +} + +func RecordsFilterByFunction(queriedFunctionName string, recorders *crd.RecorderList, triggers *crd.HTTPTriggerList) ([]byte, error) { + + // Implicit functions: + // Sometimes functions are not explicitly attached to recorders but we still want to be able to + // filter records by those functions; we do so by identifying all triggers recorders are associated with + // and checking functionReferences for those triggers. + + triggerMap := make(map[string]crd.HTTPTrigger) + for _, trigger := range triggers.Items { + triggerMap[trigger.Metadata.Name] = trigger + } + + matchingRecorders := make(map[string]bool) + + for _, recorder := range recorders.Items { + if len(recorder.Spec.Function) > 0 && recorder.Spec.Function == queriedFunctionName { + matchingRecorders[recorder.Spec.Name] = true + } else { + for _, trigger := range recorder.Spec.Triggers { + validTrigger, ok := triggerMap[trigger] + if ok { + if validTrigger.Spec.FunctionReference.Name == queriedFunctionName { + matchingRecorders[recorder.Spec.Name] = true + } + } + } + } + } + + client := NewClient() + if client == nil { + return []byte{}, errors.New("failed to create redis client") + } + + var filtered []*redisCache.RecordedEntry + + for key := range matchingRecorders { + val, err := redis.Strings(client.Do("LRANGE", key, "0", "-1")) // TODO: Prefix that distinguishes recorder lists + if err != nil { + return []byte{}, err + } + + for _, reqUID := range val { + // TODO: Check if it still exists, else clean up this value from the cache + exists, err := redis.Int(client.Do("EXISTS", reqUID)) + if err != nil { + continue + } + if exists > 0 { + val, err := redis.Bytes(client.Do("HGET", reqUID, "ReqResponse")) + if err != nil { + log.Error("Error retrieving request from Redis: ", err) + return []byte{}, err + } + entry, err := deserializeReqResponse(val, reqUID) + if err != nil { + log.Error("Error deserializing request: ", err) + return []byte{}, err + } + filtered = append(filtered, entry) + } + } + } + + resp, err := json.Marshal(filtered) + if err != nil { + return []byte{}, err + } + return resp, nil +} + +// TODO: Discuss this approach of using two different protobuf message formats +func deserializeReqResponse(value []byte, reqUID string) (*redisCache.RecordedEntry, error) { + data := &redisCache.UniqueRequest{} + err := proto.Unmarshal(value, data) + if err != nil { + log.Error("Error unmarshalling request: ", err) + return nil, err + } + log.Info("Parsed protobuf bytes: ", data) + transformed := &redisCache.RecordedEntry{ + ReqUID: reqUID, + Req: data.Req, + Resp: data.Resp, + Trigger: data.Trigger, + } + return transformed, nil +} + +func obtainInterval(from string, to string) (int64, int64, error) { + now := time.Now() + parsedFrom, err := time.ParseDuration(from) + if err != nil { + return -1, -1, err + } + + parsedTo, err := time.ParseDuration(to) + if err != nil { + return -1, -1, err + } + + then := now.Add(-1 * parsedFrom) // Start search interval + rangeStart := then.UnixNano() + + until := now.Add(-1 * parsedTo) // End search interval + rangeEnd := until.UnixNano() + + return rangeStart, rangeEnd, nil +} + +func includesTrigger(triggers []string, query string) bool { + for _, trigger := range triggers { + if trigger == query { + return true + } + } + return false +} diff --git a/redis/spec.proto b/redis/spec.proto new file mode 100644 index 00000000..4e776e4d --- /dev/null +++ b/redis/spec.proto @@ -0,0 +1,51 @@ +/* +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. +*/ + +syntax = 'proto3'; + +package redisCache; + +message Request { + string Method = 1; + map URL = 2; + map Header = 3; + string Host = 4; + map Form = 5; + map PostForm = 6; +} + +message Response { + string Status = 7; + int32 StatusCode = 8; +} + +message UniqueRequest { + Request Req = 10; + Response Resp = 11; + string Trigger = 12; +} + +// We need this because we don't store the ReqUID in the above Message (UniqueRequest) +// and we don't store the UID there because the key in Redis is the ReqUID -- including it in the value would be +// unnecessary duplication +// but we need it here because we want to display to the user the ReqResponses for a given ReqUID +message RecordedEntry { + string ReqUID = 13; + Request Req = 14; + Response Resp = 15; + string Trigger = 16; +} + diff --git a/router/functionHandler.go b/router/functionHandler.go index b08b0426..4720ab7a 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -17,20 +17,26 @@ limitations under the License. package router import ( + "bytes" "fmt" + "io/ioutil" "log" "net" "net/http" "net/http/httputil" "net/url" + "strings" "time" "github.com/gorilla/mux" + "github.com/satori/go.uuid" + "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/fission/fission" "github.com/fission/fission/crd" executorClient "github.com/fission/fission/executor/client" + "github.com/fission/fission/redis" ) type tsRoundTripperParams struct { @@ -42,10 +48,13 @@ type tsRoundTripperParams struct { type functionHandler struct { fmap *functionServiceMap + frmap *functionRecorderMap + trmap *triggerRecorderMap executor *executorClient.Client function *metav1.ObjectMeta httpTrigger *crd.HTTPTrigger tsRoundTripperParams *tsRoundTripperParams + recorderName string } // A layer on top of http.DefaultTransport, with retries. @@ -83,6 +92,27 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt var needExecutor, serviceUrlFromExecutor bool var serviceUrl *url.URL + // TODO: Keep? --> Needed for queries encoded in URL before they're stripped by the proxy + var originalUrl url.URL + originalUrl = *req.URL + + // Iff this request needs to be recorded, we save the body + var postedBody string + if len(roundTripper.funcHandler.recorderName) > 0 { + if req.ContentLength > 0 { + p := make([]byte, req.ContentLength) + buf, _ := ioutil.ReadAll(req.Body) + // We need two io readers because a single reader will drain the buffer, hence we keep a replacement copy + rdr1 := ioutil.NopCloser(bytes.NewBuffer(buf)) + rdr2 := ioutil.NopCloser(bytes.NewBuffer(buf)) + + rdr1.Read(p) + postedBody = string(p) + logrus.Info(fmt.Sprintf("%v", postedBody)) + req.Body = rdr2 + } + } + // Metrics stuff startTime := time.Now() funcMetricLabels := &functionLabels{ @@ -177,6 +207,27 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt functionCallCompleted(funcMetricLabels, httpMetricLabels, overhead, time.Since(startTime), resp.ContentLength) + // if transport.RoundTrip succeeds and it was a cached entry, then tapService + if !serviceUrlFromExecutor { + go roundTripper.funcHandler.tapService(serviceUrl) + } + + trigger := "" + if roundTripper.funcHandler.httpTrigger != nil { + trigger = roundTripper.funcHandler.httpTrigger.Metadata.Name + } else { + log.Println("No trigger attached.") // Wording? + } + + if len(roundTripper.funcHandler.recorderName) > 0 { + redis.Record( + trigger, + roundTripper.funcHandler.recorderName, + req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, roundTripper.funcHandler.function.Namespace, + time.Now().UnixNano(), + ) + } + // return response back to user return resp, nil } @@ -227,6 +278,14 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * request.Header.Add(fmt.Sprintf("X-Fission-Params-%v", k), v) } + var reqUID string + if len(fh.recorderName) > 0 { + UID := strings.ToLower(uuid.NewV4().String()) + reqUID = "REQ" + UID + request.Header.Add("X-Fission-ReqUID", reqUID) + log.Print("Record request with ReqUID: ", reqUID) + } + // system params MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request) diff --git a/router/functionRecorderMap.go b/router/functionRecorderMap.go new file mode 100644 index 00000000..ca66edd8 --- /dev/null +++ b/router/functionRecorderMap.go @@ -0,0 +1,63 @@ +/* +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 router + +import ( + "log" + "time" + + "github.com/fission/fission" + "github.com/fission/fission/cache" + "github.com/fission/fission/crd" +) + +type ( + functionRecorderMap struct { + cache *cache.Cache // map[string]*crd.Recorder + } +) + +// Why do we need an expiry? +func makeFunctionRecorderMap(expiry time.Duration) *functionRecorderMap { + return &functionRecorderMap{ + cache: cache.MakeCache(expiry, 0), + } +} + +func (frmap *functionRecorderMap) lookup(function string) (*crd.Recorder, error) { + item, err := frmap.cache.Get(function) + if err != nil { + return nil, err + } + u := item.(*crd.Recorder) + return u, nil +} + +func (frmap *functionRecorderMap) assign(function string, recorder *crd.Recorder) { + err, _ := frmap.cache.Set(function, recorder) + if err != nil { + if e, ok := err.(fission.Error); ok && e.Code == fission.ErrorNameExists { + return + } + log.Printf("error caching recorder for function name with a different value: %v", err) + } + +} + +func (frmap *functionRecorderMap) remove(function string) error { + return frmap.cache.Delete(function) +} diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 0999ed50..1f17f24f 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -49,12 +49,12 @@ type HTTPTriggerSet struct { functions []crd.Function funcStore k8sCache.Store funcController k8sCache.Controller + recorderSet *RecorderSet updateRouterRequestChannel chan struct{} - - tsRoundTripperParams *tsRoundTripperParams + tsRoundTripperParams *tsRoundTripperParams } -func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClient, +func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient, kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { httpTriggerSet := &HTTPTriggerSet{ functionServiceMap: fmap, @@ -66,8 +66,9 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClie updateRouterRequestChannel: make(chan struct{}), tsRoundTripperParams: params, } - var tStore, fnStore k8sCache.Store + var tStore, fnStore, rStore k8sCache.Store var tController, fnController k8sCache.Controller + var recorderSet *RecorderSet if httpTriggerSet.crdClient != nil { tStore, tController = httpTriggerSet.initTriggerController() httpTriggerSet.triggerStore = tStore @@ -76,6 +77,8 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClie httpTriggerSet.funcStore = fnStore httpTriggerSet.funcController = fnController } + recorderSet = MakeRecorderSet(httpTriggerSet, crdClient, rStore, frmap, trmap) + httpTriggerSet.recorderSet = recorderSet return httpTriggerSet, tStore, fnStore } @@ -92,6 +95,11 @@ func (ts *HTTPTriggerSet) subscribeRouter(ctx context.Context, mr *mutableRouter go ts.updateRouter() go ts.runWatcher(ctx, ts.funcController) go ts.runWatcher(ctx, ts.triggerController) + if ts.recorderSet.recController != nil { + go ts.runWatcher(ctx, ts.recorderSet.recController) + } else { + log.Fatal("Failed to run recorder Controller") + } } func defaultHomeHandler(w http.ResponseWriter, r *http.Request) { @@ -121,6 +129,14 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { continue } + var recorderName string + recorder, err := ts.recorderSet.triggerRecorderMap.lookup(trigger.Metadata.Name) + if err == nil && recorder != nil { + recorderName = recorder.Spec.Name + } + + //log.Printf("The trigger %v should be recorded: %v", trigger.Metadata.Name, doRecord) + if rr.resolveResultType != resolveResultSingleFunction { // not implemented yet log.Panicf("resolve result type not implemented (%v)", rr.resolveResultType) @@ -128,10 +144,13 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { fh := &functionHandler{ fmap: ts.functionServiceMap, + frmap: ts.recorderSet.functionRecorderMap, + trmap: ts.recorderSet.triggerRecorderMap, function: rr.functionMetadata, executor: ts.executor, httpTrigger: &trigger, tsRoundTripperParams: ts.tsRoundTripperParams, + recorderName: recorderName, } ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler) @@ -158,11 +177,21 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { // triggers route into these. for _, function := range ts.functions { m := function.Metadata + + var recorderName string + recorder, err := ts.recorderSet.functionRecorderMap.lookup(m.Name) + if err == nil && recorder != nil { + recorderName = recorder.Spec.Name + } + fh := &functionHandler{ fmap: ts.functionServiceMap, + frmap: ts.recorderSet.functionRecorderMap, + trmap: ts.recorderSet.triggerRecorderMap, function: &m, executor: ts.executor, tsRoundTripperParams: ts.tsRoundTripperParams, + recorderName: recorderName, } muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler) } @@ -186,11 +215,22 @@ func (ts *HTTPTriggerSet) initTriggerController() (k8sCache.Store, k8sCache.Cont trigger := obj.(*crd.HTTPTrigger) go createIngress(trigger, ts.kubeClient) ts.syncTriggers() + // Check if this trigger's function needs to be recorded + fnRef := trigger.Spec.FunctionReference.Name + recorder, err := ts.recorderSet.functionRecorderMap.lookup(fnRef) + if err == nil && recorder != nil { + if len(recorder.Spec.Triggers) == 0 { + ts.recorderSet.triggerRecorderMap.assign(trigger.Metadata.Name, recorder) + } + } else { + log.Print("Unable to lookup function in functionRecorderMap") + } }, DeleteFunc: func(obj interface{}) { ts.syncTriggers() trigger := obj.(*crd.HTTPTrigger) go deleteIngress(trigger, ts.kubeClient) + go ts.recorderSet.DeleteTriggerFromRecorderMap(trigger) }, UpdateFunc: func(oldObj interface{}, newObj interface{}) { oldTrigger := oldObj.(*crd.HTTPTrigger) @@ -216,7 +256,9 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con ts.syncTriggers() }, DeleteFunc: func(obj interface{}) { + function := obj.(*crd.Function) ts.syncTriggers() + go ts.recorderSet.DeleteFunctionFromRecorderMap(function) }, UpdateFunc: func(oldObj interface{}, newObj interface{}) { oldFn := oldObj.(*crd.Function) @@ -243,6 +285,29 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con return store, controller } +func (ts *HTTPTriggerSet) initRecorderController() (k8sCache.Store, k8sCache.Controller) { + resyncPeriod := 30 * time.Second + listWatch := k8sCache.NewListWatchFromClient(ts.crdClient, "recorders", metav1.NamespaceAll, fields.Everything()) + store, controller := k8sCache.NewInformer(listWatch, &crd.Recorder{}, resyncPeriod, + k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + recorder := obj.(*crd.Recorder) + ts.recorderSet.newRecorder(recorder) + }, + DeleteFunc: func(obj interface{}) { + recorder := obj.(*crd.Recorder) + ts.recorderSet.disableRecorder(recorder) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + oldRecorder := oldObj.(*crd.Recorder) + newRecorder := newObj.(*crd.Recorder) + ts.recorderSet.updateRecorder(oldRecorder, newRecorder) + }, + }, + ) + return store, controller +} + func (ts *HTTPTriggerSet) runWatcher(ctx context.Context, controller k8sCache.Controller) { go func() { controller.Run(ctx.Done()) diff --git a/router/recorderController.go b/router/recorderController.go new file mode 100644 index 00000000..90a6b63e --- /dev/null +++ b/router/recorderController.go @@ -0,0 +1,118 @@ +package router + +import ( + "github.com/fission/fission/crd" + log "github.com/sirupsen/logrus" + "k8s.io/client-go/rest" + k8sCache "k8s.io/client-go/tools/cache" +) + +type RecorderSet struct { + httpTriggerSet *HTTPTriggerSet + + crdClient *rest.RESTClient + + recStore k8sCache.Store + recController k8sCache.Controller + + functionRecorderMap *functionRecorderMap + triggerRecorderMap *triggerRecorderMap +} + +func MakeRecorderSet(httpTriggerSet *HTTPTriggerSet, crdClient *rest.RESTClient, rStore k8sCache.Store, frmap *functionRecorderMap, trmap *triggerRecorderMap) *RecorderSet { + recorderSet := &RecorderSet{ + httpTriggerSet: httpTriggerSet, + crdClient: crdClient, + recStore: rStore, + functionRecorderMap: frmap, + triggerRecorderMap: trmap, + } + recorderSet.recStore, recorderSet.recController = httpTriggerSet.initRecorderController() + return recorderSet +} + +// All new recorders are by default enabled +func (rs *RecorderSet) newRecorder(r *crd.Recorder) { + function := r.Spec.Function + triggers := r.Spec.Triggers + + // If triggers are not explicitly specified during the creation of this recorder, + // keep track of those associated with the function specified [implicitly added triggers] + needTrackByFunction := len(triggers) == 0 + + rs.functionRecorderMap.assign(function, r) + + if needTrackByFunction { + for _, t := range rs.httpTriggerSet.triggerStore.List() { + trigger := *t.(*crd.HTTPTrigger) + if trigger.Spec.FunctionReference.Name == function { + rs.triggerRecorderMap.assign(trigger.Metadata.Name, r) + } + } + } else { + for _, trigger := range triggers { + rs.triggerRecorderMap.assign(trigger, r) + } + } + + rs.httpTriggerSet.syncTriggers() +} + +// TODO: Delete or disable? +func (rs *RecorderSet) disableRecorder(r *crd.Recorder) { + function := r.Spec.Function + triggers := r.Spec.Triggers + + log.Info("Disabling recorder ", r.Metadata.Name) + + // Account for function + err := rs.functionRecorderMap.remove(function) + if err != nil { + log.Error("Error disabling recorder (failed to remove function from functionRecorderMap): ", err) + } + + // Account for explicitly added triggers + if len(triggers) != 0 { + for _, trigger := range triggers { + err := rs.triggerRecorderMap.remove(trigger) + if err != nil { + log.Error("Error disabling recorder (failed to remove triggers from triggerRecorderMap): ", err) + } + } + } else { + // Account for implicitly added triggers + for _, t := range rs.httpTriggerSet.triggerStore.List() { + trigger := *t.(*crd.HTTPTrigger) + if trigger.Spec.FunctionReference.Name == function { + err := rs.triggerRecorderMap.remove(trigger.Metadata.Name) + if err != nil { + log.Error("Failed to remove trigger from triggerRecorderMap: ", err) + } + } + } + } + + rs.httpTriggerSet.syncTriggers() +} + +func (rs *RecorderSet) updateRecorder(old *crd.Recorder, newer *crd.Recorder) { + if newer.Spec.Enabled == true { + rs.newRecorder(newer) // TODO: Test this + } else { + rs.disableRecorder(old) + } +} + +func (rs *RecorderSet) DeleteTriggerFromRecorderMap(trigger *crd.HTTPTrigger) { + err := rs.triggerRecorderMap.remove(trigger.Metadata.Name) + if err != nil { + log.Error("Failed to remove trigger from triggerRecorderMap: ", err) + } +} + +func (rs *RecorderSet) DeleteFunctionFromRecorderMap(function *crd.Function) { + err := rs.functionRecorderMap.remove(function.Metadata.Name) + if err != nil { + log.Error("Failed to remove function from functionRecorderMap: ", err) + } +} diff --git a/router/router.go b/router/router.go index 19ade766..4b208b27 100644 --- a/router/router.go +++ b/router/router.go @@ -86,6 +86,10 @@ func Start(port int, executorUrl string) { fmap := makeFunctionServiceMap(time.Minute) + frmap := makeFunctionRecorderMap(time.Minute) + + trmap := makeTriggerRecorderMap(time.Minute) + fissionClient, kubeClient, _, err := crd.MakeFissionClient() if err != nil { log.Fatalf("Error connecting to kubernetes API: %v", err) @@ -120,13 +124,14 @@ func Start(port int, executorUrl string) { log.Fatalf("Failed to parse max retry times: %v", err) } - triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, kubeClient, executor, restClient, + triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{ timeout: timeout, timeoutExponent: timeoutExponent, keepAlive: keepAlive, maxRetries: maxRetries, }) + resolver := makeFunctionReferenceResolver(fnStore) go serveMetric() diff --git a/router/router_test.go b/router/router_test.go index 23068903..92684301 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -58,8 +58,12 @@ func TestRouter(t *testing.T) { } frr.refCache.Set(nfr, rr) + frmap := makeFunctionRecorderMap(time.Minute) + + trmap := makeTriggerRecorderMap(time.Minute) + // HTTP trigger set with a trigger for this function - triggers, _, _ := makeHTTPTriggerSet(fmap, nil, nil, nil, nil, + triggers, _, _ := makeHTTPTriggerSet(fmap, frmap, trmap, nil, nil, nil, nil, &tsRoundTripperParams{ timeout: 50 * time.Millisecond, timeoutExponent: 2, diff --git a/router/triggerRecorderMap.go b/router/triggerRecorderMap.go new file mode 100644 index 00000000..59e164d2 --- /dev/null +++ b/router/triggerRecorderMap.go @@ -0,0 +1,62 @@ +/* +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 router + +import ( + "log" + "time" + + "github.com/fission/fission" + "github.com/fission/fission/cache" + "github.com/fission/fission/crd" +) + +type ( + triggerRecorderMap struct { + cache *cache.Cache // map[string]*crd.Recorder + } +) + +func makeTriggerRecorderMap(expiry time.Duration) *triggerRecorderMap { + return &triggerRecorderMap{ + cache: cache.MakeCache(expiry, 0), + } +} + +func (trmap *triggerRecorderMap) lookup(trigger string) (*crd.Recorder, error) { + item, err := trmap.cache.Get(trigger) + if err != nil { + return nil, err + } + u := item.(*crd.Recorder) + return u, nil +} + +func (trmap *triggerRecorderMap) assign(trigger string, recorder *crd.Recorder) { + err, _ := trmap.cache.Set(trigger, recorder) + if err != nil { + if e, ok := err.(fission.Error); ok && e.Code == fission.ErrorNameExists { + return + } + log.Printf("error caching recorder for function name with a different value: %v", err) + } + +} + +func (trmap *triggerRecorderMap) remove(trigger string) error { + return trmap.cache.Delete(trigger) +} diff --git a/test/tests/mqtrigger/test_mqtrigger.sh b/test/tests/mqtrigger/test_mqtrigger.sh deleted file mode 100755 index 73dfbdd0..00000000 --- a/test/tests/mqtrigger/test_mqtrigger.sh +++ /dev/null @@ -1,61 +0,0 @@ -#!/bin/bash - -# -# Create a function and trigger it using NATS -# - -set -euo pipefail -set +x - -ROOT=$(dirname $0)/../.. -DIR=$(dirname $0) - -clusterID="fissionMQTrigger" -topic="foo.bar" -resptopic="foo.foo" -expectedRespOutput="[foo.foo]: 'Hello, World!'" - -log "Pre-test cleanup" -fission env delete --name nodejs || true - -log "Creating nodejs env" -fission env create --name nodejs --image fission/node-env -trap "fission env delete --name nodejs" EXIT - -log "Creating function" -fn=hello-$(date +%s) -fission fn create --name $fn --env nodejs --code $DIR/main.js --method GET -trap "fission fn delete --name $fn" EXIT - -log "Creating message queue trigger" -mqt=mqt-$(date +%s) -fission mqtrigger create --name $mqt --function $fn --mqtype "nats-streaming" --topic $topic --resptopic $resptopic -trap "fission mqtrigger delete --name $mqt" EXIT - -# wait until nats trigger is created -sleep 5 - -# -# Send a message -# -log "Sending message" -go run $DIR/stan-pub.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id clientPub $topic "" - -# -# Wait for message on response topic -# -log "Waiting for response" -TIMEOUT=timeout -if [ $(uname -s) == 'Darwin' ] -then - # If this fails on mac os, do "brew install coreutils". - TIMEOUT=gtimeout -fi -response=$($TIMEOUT 120s go run $DIR/stan-sub.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id clientSub $resptopic 2>&1) - -if [[ "$response" != "$expectedRespOutput" ]]; then - log "$response is not equal to $expectedRespOutput" - exit 1 -fi - -log "Subscriber received expected response: $response" diff --git a/test/tests/recordreplay/greetings.py b/test/tests/recordreplay/greetings.py new file mode 100644 index 00000000..d2a4a8a9 --- /dev/null +++ b/test/tests/recordreplay/greetings.py @@ -0,0 +1,9 @@ +from flask import request + +def main(): + content = request.get_json(force=True) + title = content['title'] + name = content['name'] + item = content['item'] + g = "Greetings, {} {}. May I take your {}?\n".format(title, name, item) + return g diff --git a/test/tests/recordreplay/rendezvous.py b/test/tests/recordreplay/rendezvous.py new file mode 100644 index 00000000..08376aef --- /dev/null +++ b/test/tests/recordreplay/rendezvous.py @@ -0,0 +1,7 @@ +from flask import request + +def main(): + time = request.args.get('time') + date = request.args.get('date') + r = "We'll meet at {} on {}.\n".format(time, date) + return r diff --git a/test/tests/recordreplay/test_record_greetings.sh b/test/tests/recordreplay/test_record_greetings.sh new file mode 100755 index 00000000..e6557164 --- /dev/null +++ b/test/tests/recordreplay/test_record_greetings.sh @@ -0,0 +1,55 @@ +#!/bin/bash + +# +# Simple end-to-end test of record with POST +# Two recorders tested: by function, by trigger (TODO) +# + +set -euo pipefail +set +x + +ROOT=$(dirname $0)/../.. +DIR=$(dirname $0) + +echo "Pre-test cleanup" +fission env delete --name python || true + +echo "Creating python env" +fission env create --name python --image fission/python-env +# trap "fission env delete --name python" EXIT + +echo "Creating function" +fn=greetings-$(date +%s) +fission fn create --name $fn --env python --code $DIR/greetings.py --method GET +# trap "fission function delete --name $fn" EXIT + +echo "Creating http trigger" +generated=$(fission route create --function $fn --method POST --url greetings | awk '{print $2}'| tr -d "'") + +# Wait until trigger is created +sleep 5 + +echo "Creating recorder" +recName="gacrux" +fission recorder create --name $recName --function $fn +fission recorder get --name $recName +# trap "fission recorder delete --name $recName" EXIT + +# Wait until recorder is created +sleep 5 + +echo "Issuing cURL request:" +resp=$(curl -X POST "http://$FISSION_ROUTER/greetings" -d "{\"title\":\"Madam\",\"name\":\"Thanh\",\"item\":\"coat\"}") +expectedR="Greetings, Madam Thanh. May I take your coat?" +recordedStatus="$(fission records view --from 15s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedS="200OK" + +trap "fission recorder delete --name $recName && fission ht delete --name $generated && fission function delete --name $fn && fission env delete --name python" EXIT + +if [ "$resp" != "$expectedR" ] || [ "$recordedStatus" != "$expectedS" ]; then + echo "Response is not equal to expected response." + exit 1 +fi + +echo "Passed." +exit 0 \ No newline at end of file diff --git a/test/tests/recordreplay/test_record_rv.sh b/test/tests/recordreplay/test_record_rv.sh new file mode 100755 index 00000000..480d6451 --- /dev/null +++ b/test/tests/recordreplay/test_record_rv.sh @@ -0,0 +1,95 @@ +#!/bin/bash + +# +# Simple end-to-end test of record with GET +# 0) Setup: Two triggers created for a function (different urls, urlA and urlB) +# 1) One recorder created for that function, two cURL requests made to both urls, check both are recorded +# 2) One recorder created for a particular trigger (urlB), both requests repeated, check only the one for urlB was recorded +# + +set -euo pipefail +set +x + +ROOT=$(dirname $0)/../.. +DIR=$(dirname $0) +expectedR="We'll meet at 9 on Tuesday." + +echo "Pre-test cleanup" +fission env delete --name python || true + +echo "Creating python env" +fission env create --name python --image fission/python-env + +echo "Creating function" +fn=rv-$(date +%s) +fission fn create --name $fn --env python --code $DIR/rendezvous.py --method GET + +echo "Creating trigger A" +generatedA=$(fission route create --function $fn --method GET --url rvA | awk '{print $2}'| tr -d "'") + +echo "Creating trigger B" +generatedB=$(fission route create --function $fn --method GET --url rvB | awk '{print $2}'| tr -d "'") + +# Wait until triggers are created +sleep 5 + +echo "Creating recorder by function" +recName="regulus" +fission recorder create --name $recName --function $fn +fission recorder get --name $recName + +# Wait until recorder is created +sleep 5 + +echo "Issuing cURL request to urlA:" +respA=$(curl -X GET "http://$FISSION_ROUTER/rvA?time=9&date=Tuesday") +recordedStatusA="$(fission records view --from 5s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedSA="200OK" + +# Separate records +sleep 5 + +echo "Issuing cURL request to urlB:" +respB=$(curl -X GET "http://$FISSION_ROUTER/rvB?time=9&date=Tuesday") +recordedStatusB="$(fission records view --from 5s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedSB="200OK" + +if [ "$respA" != "$expectedR" ] || [ "$recordedStatusA" != "$expectedSA" ] || [ "$recordedStatusB" != "$expectedSB" ]; then + echo "Failed at test case 1." + exit 1 +fi + +echo "Test case 1) Passed." + +# Delete first recorder +fission recorder delete --name $recName + +sleep 5 + +echo "Creating recorder by trigger" +recName2="regulus2" +fission recorder create --name $recName2 --trigger $generatedB +fission recorder get --name $recName2 + +echo "Issuing cURL request to urlA:" +respA=$(curl -X GET "http://$FISSION_ROUTER/rvA?time=9&date=Tuesday") +recordedStatusA="$(fission records view --from 5s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedSA="" + +# Separate records +sleep 5 + +echo "Issuing cURL request to urlB:" +respB=$(curl -X GET "http://$FISSION_ROUTER/rvB?time=9&date=Tuesday") +recordedStatusB="$(fission records view --from 5s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedSB="200OK" + +if [ "$respA" != "$expectedR" ] || [ "$recordedStatusA" != "$expectedSA" ] || [ "$recordedStatusB" != "$expectedSB" ]; then + echo "Failed at test case 2." + exit 1 +fi + +trap "fission recorder delete --name $recName2 && fission ht delete --name $generatedA && fission ht delete --name $generatedB && fission fn delete --name $fn && fission env delete --name python" EXIT + +echo "All passed." +exit 0 \ No newline at end of file diff --git a/test/tests/recordreplay/test_recorder_update.sh b/test/tests/recordreplay/test_recorder_update.sh new file mode 100755 index 00000000..5334dc5a --- /dev/null +++ b/test/tests/recordreplay/test_recorder_update.sh @@ -0,0 +1,100 @@ +#!/bin/bash + +# +# Simple test of recorder updates +# 0) Setup: One recorder created for function with trigger +# 1) Recorder disabled, cURL request made, check not saved. +# 2) Recorder re-enabled, request repeated, check now saved. +# 3) New trigger created for same function recorded w/ different url, recorder updated to observe requests for this trigger +# Request repeated at new url, check saved. +# + +set -euo pipefail +set +x + +ROOT=$(dirname $0)/../.. +DIR=$(dirname $0) + +echo "Pre-test cleanup" +fission env delete --name python || true + +echo "Creating python env" +fission env create --name python --image fission/python-env + +echo "Creating function" +fn=rv-$(date +%s) +fission fn create --name $fn --env python --code $DIR/rendezvous.py --method GET + +echo "Creating http trigger" +generated=$(fission route create --function $fn --method GET --url rv | awk '{print $2}'| tr -d "'") + +# Wait until trigger is created +sleep 5 + +echo "Creating recorder" +recName="regulus" +fission recorder create --name $recName --function $fn +fission recorder get --name $recName + +# Wait until recorder is created +sleep 5 + +# Disable recorder +fission recorder update --name $recName --disable +sleep 5 + +echo "Issuing cURL request that should not be recorded:" +resp=$(curl -X GET "http://$FISSION_ROUTER/rv?time=9&date=Tuesday") +recordedStatus="$(fission records view --from 15s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedR="We'll meet at 9 on Tuesday." +expectedS="" + +if [ "$resp" != "$expectedR" ] || [ "$recordedStatus" != "$expectedS" ]; then + echo "Response is not equal to expected response." + exit 1 +fi + +echo "Test case 1) Passed." + +# Reenable recorder +fission recorder update --name $recName --enable +sleep 5 + +echo "Issuing cURL request that should be recorded:" +resp=$(curl -X GET "http://$FISSION_ROUTER/rv?time=9&date=Tuesday") +expectedR="We'll meet at 9 on Tuesday." +recordedStatus="$(fission records view --from 15s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedS="200OK" + +if [ "$resp" != "$expectedR" ] || [ "$recordedStatus" != "$expectedS" ]; then + echo "Response is not equal to expected response." + exit 1 +fi + +echo "Test case 2) Passed." + +# Create new trigger for same function recorded w/ different url +generated2=$(fission route create --function $fn --method GET --url rv2 | awk '{print $2}'| tr -d "'") +echo "New trigger: $generated2" + +# Update recorder to observe new trigger +fission recorder update --name $recName --trigger $generated2 +fission recorder list + +echo "Issuing cURL request that should be recorded:" +resp=$(curl -X GET "http://$FISSION_ROUTER/rv2?time=9&date=Tuesday") +expectedR="We'll meet at 9 on Tuesday." +recordedStatus="$(fission records view --from 15s --to 0s -v | awk 'FNR == 2 {print $4$5}')" +expectedS="200OK" + +if [ "$resp" != "$expectedR" ] || [ "$recordedStatus" != "$expectedS" ]; then + echo "Response is not equal to expected response." + exit 1 +fi + +echo "Test case 3) Passed." + +trap "fission recorder delete --name $recName && fission ht delete --name $generated && fission ht delete --name $generated2 && fission function delete --name $fn && fission env delete --name python" EXIT + +echo "All passed." +exit 0 \ No newline at end of file diff --git a/types.go b/types.go index 96793209..e9868132 100644 --- a/types.go +++ b/types.go @@ -51,6 +51,7 @@ type ( MessageQueueType = fv1.MessageQueueType MessageQueueTriggerSpec = fv1.MessageQueueTriggerSpec TimeTriggerSpec = fv1.TimeTriggerSpec + RecorderSpec = fv1.RecorderSpec ) type (