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

Drop unreleased features (record & replay) (#1406)

Browse Source 
1. The records are stored in redis which is not migratable to another cluster for the testing purposes.
2. Some of the requests fields are not recorded.
3. People should consider using https://github.com/buger/goreplay which is an existing mature and well-tested solution for testing purposes.
This commit is contained in:
Ta-Ching Chen
2019-11-13 15:10:32 +08:00
committed by GitHub
parent 1cda7e051b
commit cf2d35291e
51 changed files with 10 additions and 3544 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/fission.io/v1/typeDeepcopyFunc.go
-25
View File
@@ -56,10 +56,6 @@ func (c *CanaryConfig) GetObjectKind() schema.ObjectKind {
return &c.TypeMeta
}
func (r *Recorder) GetObjectKind() schema.ObjectKind {
return &r.TypeMeta
}
func (f *Function) GetObjectMeta() metav1.Object {
return &f.Metadata
}
@@ -85,10 +81,6 @@ func (c *CanaryConfig) GetObjectMeta() metav1.Object {
return &c.Metadata
}
func (r *Recorder) GetObjectMeta() metav1.Object {
return &r.Metadata
}
func (fl *FunctionList) GetObjectKind() schema.ObjectKind {
return &fl.TypeMeta
}
@@ -110,9 +102,6 @@ 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 (cl *CanaryConfigList) GetObjectKind() schema.ObjectKind {
return &cl.TypeMeta
@@ -140,10 +129,6 @@ func (pl *PackageList) GetListMeta() metav1.ListInterface {
return &pl.Metadata
}
func (rl *RecorderList) GetListMeta() metav1.ListInterface {
return &rl.Metadata
}
func (cl *CanaryConfigList) GetListMeta() metav1.ListInterface {
return &cl.Metadata
}
@@ -279,13 +264,3 @@ func (ml *MessageQueueTriggerList) Validate() error {
}
return result.ErrorOrNil()
}
func (r *Recorder) Validate() error {
result := &multierror.Error{}
result = multierror.Append(result,
validateMetadata("Recorder", r.Metadata),
r.Spec.Validate())
return result.ErrorOrNil()
}
pkg/apis/fission.io/v1/types.go
-33
View File
@@ -157,23 +157,6 @@ type (
Items []MessageQueueTrigger `json:"items"`
}
// Recorder allows user to record all traffic payload to a certain function.
// +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"`
}
// RecorderList is a list of Recorders.
// +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"`
}
// CanaryConfig is for canary deployment of two functions.
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
CanaryConfig struct {
@@ -642,22 +625,6 @@ 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 of recorder resource
Name string `json:"name"`
// Function to collect requests/responses
Function string `json:"function"`
// HTTP trigger to record the requests and responses.
Triggers []string `json:"triggers"`
RetentionPolicy string `json:"-"` // `json:"retentionPolicy"`
EvictionPolicy string `json:"-"` // `json:"evictionPolicy"`
Enabled bool `json:"enabled"`
}
// TimeTrigger invokes the specific function at a time or
// times specified by a cron string.
TimeTriggerSpec struct {
pkg/apis/fission.io/v1/validation.go
-27
View File
@@ -544,33 +544,6 @@ func (spec MessageQueueTriggerSpec) Validate() error {
return result.ErrorOrNil()
}
func (spec RecorderSpec) Validate() error {
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 {
result := &multierror.Error{}
pkg/apis/fission.io/v1/zz_generated.deepcopy.go
-81
View File
@@ -801,87 +801,6 @@ 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
pkg/controller/api.go
-13
View File
@@ -239,19 +239,6 @@ func (api *API) GetHandler() http.Handler {
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/replay/{reqUID}", api.ReplayByReqUID).Methods("GET")
r.HandleFunc("/v2/secrets/{secret}", api.SecretGet).Methods("GET")
r.HandleFunc("/v2/configmaps/{configmap}", api.ConfigMapGet).Methods("GET")
pkg/controller/client/recorder.go
-236
View File
@@ -1,236 +0,0 @@
/*
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 (
"encoding/json"
"fmt"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/redis/build/gen"
)
func (c *Client) RecorderCreate(r *fv1.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 := c.create("recorders", "application/json", 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) (*fv1.Recorder, error) {
relativeUrl := fmt.Sprintf("recorders/%v", m.Name)
relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace)
resp, err := c.get(relativeUrl)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := c.handleResponse(resp)
if err != nil {
return nil, err
}
var r fv1.Recorder
err = json.Unmarshal(body, &r)
if err != nil {
return nil, err
}
return &r, nil
}
func (c *Client) RecorderUpdate(recorder *fv1.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) ([]fv1.Recorder, error) {
relativeUrl := "recorders"
resp, err := c.get(relativeUrl)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := c.handleResponse(resp)
if err != nil {
return nil, err
}
recorders := make([]fv1.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 := c.get(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 := c.get(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 := c.get(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 := c.get(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
}
pkg/controller/client/replayer.go
-46
View File
@@ -1,46 +0,0 @@
/*
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 (
"encoding/json"
"fmt"
)
func (c *Client) ReplayByReqUID(reqUID string) ([]string, error) {
relativeUrl := fmt.Sprintf("replay/%v", reqUID)
resp, err := c.get(relativeUrl)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := c.handleResponse(resp)
if err != nil {
return nil, err
}
replayed := make([]string, 0)
err = json.Unmarshal(body, &replayed)
if err != nil {
return nil, err
}
return replayed, nil
}
pkg/controller/recorderApi.go
-148
View File
@@ -1,148 +0,0 @@
/*
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"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
ferror "github.com/fission/fission/pkg/error"
)
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 fv1.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 fv1.Recorder
err = json.Unmarshal(body, &recorder)
if err != nil {
a.respondWithError(w, err)
return
}
if name != recorder.Metadata.Name {
err = ferror.MakeError(ferror.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(""))
}
pkg/controller/recordsApi.go
-95
View File
@@ -1,95 +0,0 @@
/*
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/pkg/redis"
)
func (a *API) RecordsApiListAll(w http.ResponseWriter, r *http.Request) {
resp, err := redis.RecordsListAll(a.logger.Named("redis"))
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(a.logger.Named("redis"), 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(a.logger.Named("redis"), 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(a.logger.Named("redis"), from, to)
if err != nil {
a.respondWithError(w, err)
return
}
a.respondWithSuccess(w, resp)
}
pkg/controller/replayAPI.go
-40
View File
@@ -1,40 +0,0 @@
/*
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 (
"fmt"
"net/http"
"github.com/gorilla/mux"
"github.com/fission/fission/pkg/redis"
)
func (a *API) ReplayByReqUID(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
queriedID := vars["reqUID"]
routerUrl := fmt.Sprintf("http://router.%v", podNamespace)
resp, err := redis.ReplayByReqUID(a.logger, routerUrl, queriedID)
if err != nil {
a.respondWithError(w, err)
return
}
a.respondWithSuccess(w, resp)
}
pkg/crd/client.go
-10
View File
@@ -155,13 +155,6 @@ func configureClient(config *rest.Config) {
&metav1.ListOptions{},
&metav1.DeleteOptions{},
)
scheme.AddKnownTypes(
groupversion,
&fv1.Recorder{},
&fv1.RecorderList{},
&metav1.ListOptions{},
&metav1.DeleteOptions{},
)
scheme.AddKnownTypes(
groupversion,
&fv1.CanaryConfig{},
@@ -224,9 +217,6 @@ 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)
}
pkg/crd/crd.go
-16
View File
@@ -157,22 +157,6 @@ func EnsureFissionCRDs(logger *zap.Logger, clientset *apiextensionsclient.Client
},
},
},
// 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{
pkg/crd/recorder.go
-122
View File
@@ -1,122 +0,0 @@
/*
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"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
)
type (
RecorderInterface interface {
Create(*fv1.Recorder) (*fv1.Recorder, error)
Get(name string) (*fv1.Recorder, error)
Update(*fv1.Recorder) (*fv1.Recorder, error)
Delete(name string, opts *metav1.DeleteOptions) error
List(opts metav1.ListOptions) (*fv1.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 *fv1.Recorder) (*fv1.Recorder, error) {
var result fv1.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) (*fv1.Recorder, error) {
var result fv1.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 *fv1.Recorder) (*fv1.Recorder, error) {
var result fv1.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) (*fv1.RecorderList, error) {
var result fv1.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()
}
pkg/fission-cli/cmd/recorder/command.go
-83
View File
@@ -1,83 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"github.com/spf13/cobra"
wrapper "github.com/fission/fission/pkg/fission-cli/cliwrapper/driver/cobra"
"github.com/fission/fission/pkg/fission-cli/flag"
)
func Commands() *cobra.Command {
createCmd := &cobra.Command{
Use: "create",
Short: "Create a recorder",
RunE: wrapper.Wrapper(Create),
}
wrapper.SetFlags(createCmd, flag.FlagSet{
Optional: []flag.Flag{flag.RecorderName, flag.RecorderFn, flag.RecorderTriggers, flag.SpecSave},
})
getCmd := &cobra.Command{
Use: "get",
Short: "Get recorder details",
RunE: wrapper.Wrapper(Get),
}
wrapper.SetFlags(getCmd, flag.FlagSet{
Required: []flag.Flag{flag.RecorderName},
})
updateCmd := &cobra.Command{
Use: "update",
Short: "Update a recorder",
RunE: wrapper.Wrapper(Update),
}
wrapper.SetFlags(getCmd, flag.FlagSet{
Required: []flag.Flag{flag.RecorderName},
Optional: []flag.Flag{flag.RecorderFn, flag.RecorderTriggers, flag.RecorderEnabled, flag.RecorderDisabled},
})
deleteCmd := &cobra.Command{
Use: "delete",
Short: "Delete a recorder",
RunE: wrapper.Wrapper(Delete),
}
wrapper.SetFlags(deleteCmd, flag.FlagSet{
Required: []flag.Flag{flag.RecorderName},
Optional: []flag.Flag{flag.NamespaceRecorder},
})
listCmd := &cobra.Command{
Use: "list",
Short: "List all recorders in a namespace if specified, else, list recorders across all namespaces",
RunE: wrapper.Wrapper(List),
}
wrapper.SetFlags(deleteCmd, flag.FlagSet{
Optional: []flag.Flag{flag.NamespaceRecorder},
})
command := &cobra.Command{
Use: "recorder",
Short: "Create, update and manage recorders",
Hidden: true,
}
command.AddCommand(createCmd, getCmd, updateCmd, deleteCmd, listCmd)
return command
}
pkg/fission-cli/cmd/recorder/create.go
-124
View File
@@ -1,124 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"fmt"
"strings"
"github.com/pkg/errors"
"github.com/satori/go.uuid"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/cmd/spec"
"github.com/fission/fission/pkg/fission-cli/util"
)
type CreateSubCommand struct {
client *client.Client
recorder *fv1.Recorder
}
func Create(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := CreateSubCommand{
client: c,
}
return opts.do(input)
}
func (opts *CreateSubCommand) do(input cli.Input) error {
err := opts.complete(input)
if err != nil {
return err
}
return opts.run(input)
}
func (opts *CreateSubCommand) complete(input cli.Input) error {
recName := input.String("name")
if len(recName) == 0 {
recName = uuid.NewV4().String()
}
fnName := input.String("function")
triggersOriginal := input.StringSlice("trigger")
// Function XOR triggers can be given
if len(fnName) == 0 && len(triggersOriginal) == 0 {
return errors.New("Need to specify at least one function or one trigger, use --function, --trigger")
}
if len(fnName) != 0 && len(triggersOriginal) != 0 {
return errors.New("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 {
if len(name) > 0 {
triggers = append(triggers, name)
}
}
}
// TODO: Define appropriate set of policies and defaults
//retPolicy := flags.String("retention")
//evictPolicy := flags.String("eviction")
opts.recorder = &fv1.Recorder{
Metadata: metav1.ObjectMeta{
Name: recName,
Namespace: "default",
},
Spec: fv1.RecorderSpec{
Name: recName,
Function: fnName,
Triggers: triggers,
RetentionPolicy: "Permanent", // TODO: Implement customizable policies for expiration of records
EvictionPolicy: "None",
Enabled: true,
},
}
return nil
}
func (opts *CreateSubCommand) run(input cli.Input) error {
// If we're writing a spec, don't call the API
if input.Bool("spec") {
specFile := fmt.Sprintf("recorder-%v.yaml", opts.recorder.Metadata.Name)
err := spec.SpecSave(*opts.recorder, specFile)
if err != nil {
return errors.Wrap(err, "error creating recorder spec")
}
return nil
}
_, err := opts.client.RecorderCreate(opts.recorder)
if err != nil {
return errors.Wrap(err, "error creating recorder")
}
fmt.Printf("recorder '%s' created\n", opts.recorder.Metadata.Name)
return nil
}
pkg/fission-cli/cmd/recorder/delete.go
-69
View File
@@ -1,69 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"fmt"
"github.com/pkg/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
)
type DeleteSubCommand struct {
client *client.Client
metadata *metav1.ObjectMeta
}
func Delete(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := DeleteSubCommand{
client: c,
}
return opts.do(input)
}
func (opts *DeleteSubCommand) do(input cli.Input) error {
err := opts.complete(input)
if err != nil {
return err
}
return opts.run(input)
}
func (opts *DeleteSubCommand) complete(input cli.Input) error {
opts.metadata = &metav1.ObjectMeta{
Name: input.String("name"),
Namespace: input.String("recorderns"),
}
return nil
}
func (opts *DeleteSubCommand) run(input cli.Input) error {
err := opts.client.RecorderDelete(opts.metadata)
if err != nil {
return errors.Wrap(err, "error deleting recorder")
}
fmt.Printf("recorder '%v' deleted\n", opts.metadata.Name)
return nil
}
pkg/fission-cli/cmd/recorder/get.go
-82
View File
@@ -1,82 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"fmt"
"os"
"text/tabwriter"
"github.com/pkg/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
)
type GetSubCommand struct {
client *client.Client
name string
}
func Get(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := GetSubCommand{
client: c,
}
return opts.do(input)
}
func (opts *GetSubCommand) do(input cli.Input) error {
err := opts.complete(input)
if err != nil {
return err
}
return opts.run(input)
}
func (opts *GetSubCommand) complete(input cli.Input) error {
opts.name = input.String("name")
if len(opts.name) <= 0 {
return errors.New("need a recorder name, use --name")
}
return nil
}
func (opts *GetSubCommand) run(input cli.Input) error {
recorder, err := opts.client.RecorderGet(&metav1.ObjectMeta{
Name: opts.name,
Namespace: "default",
})
if err != nil {
return errors.Wrap(err, "error getting 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
}
pkg/fission-cli/cmd/recorder/list.go
-66
View File
@@ -1,66 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"fmt"
"os"
"text/tabwriter"
"github.com/pkg/errors"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
)
type ListSubCommand struct {
client *client.Client
}
func List(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := ListSubCommand{
client: c,
}
return opts.do(input)
}
func (opts *ListSubCommand) do(input cli.Input) error {
return opts.run(input)
}
func (opts *ListSubCommand) run(input cli.Input) error {
recorders, err := opts.client.RecorderList("default")
if err != nil {
return errors.Wrap(err, "error listing 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
}
pkg/fission-cli/cmd/recorder/update.go
-145
View File
@@ -1,145 +0,0 @@
/*
Copyright 2019 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 recorder
import (
"fmt"
"strings"
"github.com/pkg/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
)
type UpdateSubCommand struct {
client *client.Client
recorder *fv1.Recorder
}
func Update(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := UpdateSubCommand{
client: c,
}
return opts.do(input)
}
func (opts *UpdateSubCommand) do(input cli.Input) error {
err := opts.complete(input)
if err != nil {
return err
}
return opts.run(input)
}
func (opts *UpdateSubCommand) complete(input cli.Input) error {
recName := input.String("name")
enable := input.Bool("enable")
disable := input.Bool("disable")
//retPolicy := flags.String("retention")
//evictPolicy := flags.String("eviction")
triggers := input.StringSlice("trigger")
function := input.String("function")
if enable && disable {
return errors.New("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 {
return errors.New("Enabling or disabling a recorder with other (non-name) flags set is not supported.")
}
} else if len(triggers) == 0 && len(function) == 0 {
return errors.New("Need to specify either a function or trigger(s) for this recorder")
}
if len(recName) == 0 {
return errors.New("Need name of recorder, use --name")
}
recorder, err := opts.client.RecorderGet(&metav1.ObjectMeta{
Name: recName,
Namespace: "default",
})
if err != nil {
return errors.Wrap(err, "error getting recorder")
}
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 {
if len(name) > 0 {
newTriggers = append(newTriggers, name)
}
}
recorder.Spec.Triggers = newTriggers
updated = true
}
if len(function) > 0 {
recorder.Spec.Function = function
updated = true
}
if !updated {
return errors.New("Nothing to update. Use --function, --triggers, --enable or --disable")
}
opts.recorder = recorder
return nil
}
func (opts *UpdateSubCommand) run(input cli.Input) error {
_, err := opts.client.RecorderUpdate(opts.recorder)
if err != nil {
return errors.Wrap(err, "error updating recorder")
}
fmt.Printf("recorder '%v' updated\n", opts.recorder.Metadata.Name)
return nil
}
pkg/fission-cli/cmd/records/command.go
-47
View File
@@ -1,47 +0,0 @@
/*
Copyright 2019 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 records
import (
"github.com/spf13/cobra"
wrapper "github.com/fission/fission/pkg/fission-cli/cliwrapper/driver/cobra"
"github.com/fission/fission/pkg/fission-cli/flag"
)
func Commands() *cobra.Command {
viewCmd := &cobra.Command{
Use: "view",
Short: "View existing records",
RunE: wrapper.Wrapper(View),
}
wrapper.SetFlags(viewCmd, flag.FlagSet{
Optional: []flag.Flag{flag.RecordsFilterTimeTo, flag.RecordsFilterTimeFrom,
flag.RecordsFilterFunction, flag.RecordsFilterTrigger, flag.RecordsVerbosity,
flag.RecordsVv},
})
command := &cobra.Command{
Use: "records",
Short: "View records with optional filters",
Hidden: true,
}
command.AddCommand(viewCmd)
return command
}
pkg/fission-cli/cmd/records/view.go
-159
View File
@@ -1,159 +0,0 @@
/*
Copyright 2019 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 records
import (
"fmt"
"os"
"text/tabwriter"
"github.com/pkg/errors"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
redisCache "github.com/fission/fission/pkg/redis/build/gen"
)
type ViewSubCommand struct {
client *client.Client
}
func View(flaginput cli.Input) error {
c, err := util.GetServer(flaginput)
if err != nil {
return err
}
opts := ViewSubCommand{
client: c,
}
return opts.do(flaginput)
}
func (opts *ViewSubCommand) do(input cli.Input) error {
return opts.run(input)
}
func (opts *ViewSubCommand) run(input cli.Input) error {
var verbosity int
if input.Bool("v") && input.Bool("vv") {
return errors.New("conflicting verbosity levels, use either --v or --vv")
}
if input.Bool("v") {
verbosity = 1
}
if input.Bool("vv") {
verbosity = 2
}
function := input.String("function")
trigger := input.String("trigger")
from := input.String("from")
to := input.String("to")
//Refuse multiple filters for now
if multipleFiltersSpecified(function, trigger, from+to) {
return errors.New("maximum of one filter is currently supported, either --function, --trigger, or --from,--to")
}
if len(function) != 0 {
return recordsByFunction(opts.client, function, verbosity)
}
if len(trigger) != 0 {
return recordsByTrigger(opts.client, trigger, verbosity)
}
if len(from) != 0 && len(to) != 0 {
return recordsByTime(opts.client, from, to, verbosity)
}
err := recordsAll(opts.client, verbosity)
if err != nil {
return errors.Wrap(err, "error viewing records")
}
return nil
}
func recordsAll(client *client.Client, verbosity int) error {
records, err := client.RecordsAll()
if err != nil {
return errors.Wrap(err, "error viewing records")
}
showRecords(records, verbosity)
return nil
}
func recordsByTrigger(client *client.Client, trigger string, verbosity int) error {
records, err := client.RecordsByTrigger(trigger)
if err != nil {
return errors.Wrap(err, "error viewing records")
}
showRecords(records, verbosity)
return nil
}
// TODO: More accurate function name (function filter)
func recordsByFunction(client *client.Client, function string, verbosity int) error {
records, err := client.RecordsByFunction(function)
if err != nil {
return errors.Wrap(err, "error viewing records")
}
showRecords(records, verbosity)
return nil
}
func recordsByTime(client *client.Client, from string, to string, verbosity int) error {
records, err := client.RecordsByTime(from, to)
if err != nil {
return errors.Wrap(err, "error viewing 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
}
pkg/fission-cli/cmd/replay/command.go
-45
View File
@@ -1,45 +0,0 @@
/*
Copyright 2019 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 replay
import (
"github.com/spf13/cobra"
wrapper "github.com/fission/fission/pkg/fission-cli/cliwrapper/driver/cobra"
"github.com/fission/fission/pkg/fission-cli/flag"
)
func Commands() *cobra.Command {
replayCmd := &cobra.Command{
Use: "create",
Short: "Create a recorder",
RunE: wrapper.Wrapper(Replay),
}
wrapper.SetFlags(replayCmd, flag.FlagSet{
Required: []flag.Flag{flag.RecordsReqID},
})
command := &cobra.Command{
Use: "replay",
Short: "Replay records",
Hidden: true,
}
command.AddCommand(replayCmd)
return command
}
pkg/fission-cli/cmd/replay/replay.go
-71
View File
@@ -1,71 +0,0 @@
/*
Copyright 2019 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 replay
import (
"fmt"
"os"
"text/tabwriter"
"github.com/pkg/errors"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/util"
)
type ReplaySubCommand struct {
client *client.Client
}
func Replay(input cli.Input) error {
c, err := util.GetServer(input)
if err != nil {
return err
}
opts := ReplaySubCommand{
client: c,
}
return opts.do(input)
}
func (opts *ReplaySubCommand) do(input cli.Input) error {
return opts.run(input)
}
func (opts *ReplaySubCommand) run(input cli.Input) error {
reqUID := input.String("reqUID")
if len(reqUID) == 0 {
return errors.New("Need a reqUID, use --reqUID flag to specify")
}
responses, err := opts.client.ReplayByReqUID(reqUID)
if err != nil {
return errors.Wrap(err, "error replaying records")
}
w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
for _, resp := range responses {
fmt.Fprintf(w, "%v",
resp,
)
}
w.Flush()
return nil
}
pkg/fission-cli/cmd/spec/spec.go
-4
View File
@@ -209,10 +209,6 @@ func SpecSave(resource interface{}, specFile string) error {
typedres.TypeMeta.APIVersion = fv1.CRD_VERSION
typedres.TypeMeta.Kind = "TimeTrigger"
data, err = yaml.Marshal(typedres)
case fv1.Recorder:
typedres.TypeMeta.APIVersion = fv1.CRD_VERSION
typedres.TypeMeta.Kind = "Recorder"
data, err = yaml.Marshal(typedres)
default:
return fmt.Errorf("can't save resource %#v", resource)
}
pkg/fission-cli/cmd/support/dump.go
+5 -5
View File
@@ -90,15 +90,15 @@ func (opts *DumpSubCommand) do(input cli.Input) error {
// fission component logs & spec
"fission-components-svc-spec": resources.NewKubernetesObjectDumper(k8sClient, resources.KubernetesService,
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, redis, router, storagesvc, timer)"),
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, router, storagesvc, timer)"),
"fission-components-deployment-spec": resources.NewKubernetesObjectDumper(k8sClient, resources.KubernetesDeployment,
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, redis, router, storagesvc, timer)"),
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, router, storagesvc, timer)"),
"fission-components-daemonset-spec": resources.NewKubernetesObjectDumper(k8sClient, resources.KubernetesDaemonSet,
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, redis, router, storagesvc, timer)"),
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, router, storagesvc, timer)"),
"fission-components-pod-spec": resources.NewKubernetesObjectDumper(k8sClient, resources.KubernetesPod,
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, redis, router, storagesvc, timer)"),
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, router, storagesvc, timer)"),
"fission-components-pod-log": resources.NewKubernetesPodLogDumper(k8sClient,
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, redis, router, storagesvc, timer)"),
"svc in (buildermgr, controller, executor, influxdb, kubewatcher, logger, mqtrigger, nats-streaming, router, storagesvc, timer)"),
// fission builder logs & spec
"fission-builder-svc-spec": resources.NewKubernetesObjectDumper(k8sClient, resources.KubernetesService, "owner=buildermgr"),
pkg/fission-cli/flag/flag.go
-17
View File
@@ -74,7 +74,6 @@ var (
NamespaceEnvironment = Flag{Type: String, Name: flagkey.NamespaceEnvironment, Aliases: []string{"envns"}, Usage: "Namespace for environment object", DefaultValue: metav1.NamespaceDefault}
NamespacePackage = Flag{Type: String, Name: flagkey.NamespacePackage, Aliases: []string{"pkgns"}, Usage: "Namespace for package object", DefaultValue: metav1.NamespaceDefault}
NamespaceTrigger = Flag{Type: String, Name: flagkey.NamespaceTrigger, Aliases: []string{"triggerns"}, Usage: "Namespace for trigger object", DefaultValue: metav1.NamespaceDefault}
NamespaceRecorder = Flag{Type: String, Name: flagkey.NamespaceRecorder, Aliases: []string{"recorderns"}, Usage: "Namespace for recorder object", DefaultValue: metav1.NamespaceDefault}
NamespaceCanary = Flag{Type: String, Name: flagkey.NamespaceCanary, Aliases: []string{"canaryns"}, Usage: "Namespace for canary config object", DefaultValue: metav1.NamespaceDefault}
RunTimeMinCPU = Flag{Type: Int, Name: flagkey.RuntimeMincpu, Usage: "Minimum CPU to be assigned to pod (In millicore, minimum 1)"}
@@ -135,22 +134,6 @@ var (
MqtMaxRetries = Flag{Type: Int, Name: flagkey.MqtMaxRetries, Usage: "Maximum number of times the function will be retried upon failure", DefaultValue: 0}
MqtMsgContentType = Flag{Type: String, Name: flagkey.MqtMsgContentType, Short: "c", Usage: "Content type of messages that publish to the topic", DefaultValue: "application/json"}
RecorderName = Flag{Type: String, Name: flagkey.RecorderName, Usage: "Recorder name"}
RecorderFn = Flag{Type: String, Name: flagkey.RecorderFn, Usage: "Record Function name(s): --function=fnA"}
RecorderTriggers = Flag{Type: StringSlice, Name: flagkey.RecorderTriggers, Usage: "Record Trigger name(s): --trigger=trigger1,trigger2,trigger3"}
RecorderRetentionPolicy = Flag{Type: String, Name: flagkey.RecorderRetentionPolicy, Usage: "Retention policy (number of days)"}
RecorderEvictionPolicy = Flag{Type: String, Name: flagkey.RecorderEvictionPolcy, Usage: "Eviction policy (default LRU)"}
RecorderEnabled = Flag{Type: Bool, Name: flagkey.RecorderEnabled, Usage: "Enable recorder"}
RecorderDisabled = Flag{Type: Bool, Name: flagkey.RecorderDisabled, Usage: "Disable recorder"}
RecordsFilterTimeFrom = Flag{Type: String, Name: flagkey.RecordsFilterTimeFrom, Usage: "Filter records by time interval; specify start of interval"}
RecordsFilterTimeTo = Flag{Type: String, Name: flagkey.RecordsFilterTimeTo, Usage: "Filter records by time interval; specify end of interval"}
RecordsFilterFunction = Flag{Type: String, Name: flagkey.RecordsFilterFunction, Usage: "Filter records by function"}
RecordsFilterTrigger = Flag{Type: String, Name: flagkey.RecordsFilterTrigger, Usage: "Filter records by trigger"}
RecordsVerbosity = Flag{Type: Bool, Name: flagkey.RecordsVerbosity, Usage: "Toggle verbosity -- view more detailed requests/responses"}
RecordsVv = Flag{Type: Bool, Name: flagkey.RecordsVv, Usage: "Toggle verbosity -- view raw requests/responses"}
RecordsReqID = Flag{Type: String, Name: flagkey.RecordsReqID, Usage: "Replay a particular request by providing the reqUID (to view reqUIDs, do 'fission records view')"}
EnvName = Flag{Type: String, Name: flagkey.EnvName, Usage: "Environment name"}
EnvPoolsize = Flag{Type: Int, Name: flagkey.EnvPoolsize, Usage: "Size of the pool", DefaultValue: 3}
EnvImage = Flag{Type: String, Name: flagkey.EnvImage, Usage: "Environment image URL"}
pkg/fission-cli/flag/key/key.go
-16
View File
@@ -28,7 +28,6 @@ const (
NamespaceEnvironment = "envNamespace"
NamespacePackage = "pkgNamespace"
NamespaceTrigger = "triggerNamespace"
NamespaceRecorder = "recorderNamespace"
NamespaceCanary = "canaryNamespace"
RuntimeMincpu = "mincpu"
@@ -89,21 +88,6 @@ const (
MqtMaxRetries = "maxretries"
MqtMsgContentType = "contenttype"
RecorderName = resourceName
RecorderFn = "function"
RecorderTriggers = "trigger"
RecorderRetentionPolicy = "retention"
RecorderEvictionPolcy = "eviction"
RecorderEnabled = "enable"
RecorderDisabled = "disable"
RecordsFilterTimeFrom = "from"
RecordsFilterTimeTo = "to"
RecordsFilterFunction = "function"
RecordsFilterTrigger = "trigger"
RecordsVerbosity = "v"
RecordsVv = "vv"
RecordsReqID = "reqUID"
EnvName = resourceName
EnvPoolsize = "poolsize"
EnvImage = "image"
pkg/redis/README.md
-5
View File
@@ -1,5 +0,0 @@
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`
pkg/redis/build/gen/spec.pb.go
-305
View File
@@ -1,305 +0,0 @@
// 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,
}
pkg/redis/redis.go
-126
View File
@@ -1,126 +0,0 @@
/*
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"
"github.com/pkg/errors"
"go.uber.org/zap"
"github.com/fission/fission/pkg/redis/build/gen"
)
func NewClient() (redis.Conn, error) {
redisIP := os.Getenv("REDIS_SERVICE_HOST") // TODO: Do this here or somewhere earlier?
redisPort := os.Getenv("REDIS_SERVICE_PORT")
if len(redisIP) == 0 || len(redisPort) == 0 {
return nil, errors.New("redis host or port not supplied")
}
redisURL := fmt.Sprintf("%s:%s", redisIP, redisPort)
c, err := redis.Dial("tcp", redisURL)
if err != nil {
return nil, errors.Wrapf(err, "could not connect to Redis at url %q", redisURL)
}
return c, nil
}
func Record(logger *zap.Logger, 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, err := NewClient()
if err != nil {
logger.Error("could not create redis client", zap.Error(err))
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,
}
logger.Info("storing request", zap.Any("request", req))
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 {
logger.Error("error marshalling request", zap.Error(err))
return
}
_, err = client.Do("HMSET", reqUID, "ReqResponse", data, "Timestamp", timestamp, "Trigger", triggerName)
if err != nil {
logger.Error("error saving request", zap.Error(err))
return
}
_, err = client.Do("LPUSH", recorderName, reqUID)
if err != nil {
logger.Error("error saving recorder-request pair", zap.Error(err))
return
}
}
pkg/redis/redisApi.go
-423
View File
@@ -1,423 +0,0 @@
/*
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 (
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"strconv"
"strings"
"time"
"go.uber.org/zap"
"github.com/golang/protobuf/proto"
"github.com/gomodule/redigo/redis"
"github.com/pkg/errors"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/redis/build/gen"
)
func RecordsListAll(logger *zap.Logger) ([]byte, error) {
client, err := NewClient()
if err != nil {
return nil, errors.Wrap(err, "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 nil, 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 {
logger.Error("error retrieving request from redis", zap.Error(err))
return nil, err
}
entry, err := deserializeReqResponse(val, key)
if err != nil {
logger.Error("error deserializing request from redis", zap.Error(err))
return nil, err
}
filtered = append(filtered, entry)
}
}
if iter == 0 {
break
}
}
resp, err := json.Marshal(filtered)
if err != nil {
return nil, 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(logger *zap.Logger, from string, to string) ([]byte, error) {
rangeStart, rangeEnd, err := obtainInterval(from, to)
if err != nil {
return nil, err
}
logger.Debug("interval inferred", zap.Int64("range_start", rangeStart), zap.Int64("range_end", rangeEnd))
if rangeStart >= rangeEnd {
e := "invalid chronology - start is greater than or equal to end"
logger.Error(e, zap.Int64("range_start", rangeStart), zap.Int64("range_end", rangeEnd))
return nil, errors.New(e)
}
client, err := NewClient()
if client == nil {
return nil, errors.Wrap(err, "failed to create redis client")
}
iter := 0
var filtered []*redisCache.RecordedEntry
for {
arr, err := redis.Values(client.Do("SCAN", iter))
if err != nil {
return nil, 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 {
logger.Error("error retrieving timestamp from redis", zap.Error(err))
return nil, err
}
tsO, err := strconv.Atoi(val[0])
if err != nil {
logger.Error("error converting timestamp to int", zap.Error(err))
return nil, err
}
ts := int64(tsO)
if ts >= rangeStart && ts <= rangeEnd {
val2, err := redis.Bytes(client.Do("HGET", key, "ReqResponse"))
if err != nil {
logger.Error("error retrieving request from redis", zap.Error(err))
return nil, err
}
entry, err := deserializeReqResponse(val2, key)
if err != nil {
logger.Error("error deserializing request from redis", zap.Error(err))
return nil, err
}
filtered = append(filtered, entry)
}
}
}
if iter == 0 {
break
}
}
resp, err := json.Marshal(filtered)
if err != nil {
return nil, err
}
return resp, nil
}
func RecordsFilterByTrigger(logger *zap.Logger, queriedTriggerName string, recorders *fv1.RecorderList, triggers *fv1.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, err := NewClient()
if err != nil {
return nil, errors.Wrap(err, "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 nil, err
}
for _, reqUID := range val {
val, err := redis.Strings(client.Do("HMGET", reqUID, "Trigger")) // 1-to-1 reqUID - trigger?
if err != nil {
logger.Error("error retrieving trigger for a request from redis", zap.Error(err))
return nil, err
}
if val[0] == queriedTriggerName {
// TODO: Reconsider multiple commands
val, err := redis.Bytes(client.Do("HGET", reqUID, "ReqResponse"))
if err != nil {
logger.Error("error retrieving request from redis", zap.Error(err))
return nil, err
}
entry, err := deserializeReqResponse(val, reqUID)
if err != nil {
logger.Error("error deserializing request from redis", zap.Error(err))
return nil, err
}
filtered = append(filtered, entry)
}
}
}
resp, err := json.Marshal(filtered)
if err != nil {
return nil, err
}
return resp, nil
}
func RecordsFilterByFunction(logger *zap.Logger, queriedFunctionName string, recorders *fv1.RecorderList, triggers *fv1.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]fv1.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, err := NewClient()
if err != nil {
return nil, errors.Wrap(err, "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 nil, 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 {
logger.Error("error retrieving request from redis", zap.Error(err))
return nil, err
}
entry, err := deserializeReqResponse(val, reqUID)
if err != nil {
logger.Error("error deserializing request from redis", zap.Error(err))
return nil, err
}
filtered = append(filtered, entry)
}
}
}
resp, err := json.Marshal(filtered)
if err != nil {
return nil, 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 {
return nil, errors.Wrap(err, "error unmarshalling request")
}
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
}
func ReplayByReqUID(logger *zap.Logger, routerUrl string, queriedID string) ([]byte, error) {
client, err := NewClient()
if err != nil {
return nil, errors.Wrap(err, "failed to create redis client")
}
exists, err := redis.Int(client.Do("EXISTS", queriedID))
if exists != 1 || err != nil {
logger.Error("could not find request to replay in redis", zap.Error(err))
return nil, err
}
val, err := redis.Bytes(client.Do("HGET", queriedID, "ReqResponse"))
if err != nil {
logger.Error("could not obtain ReqResponse for ID from redis", zap.Error(err), zap.String("id", queriedID))
return nil, err
}
entry, err := deserializeReqResponse(val, queriedID)
if err != nil {
logger.Error("error deserializing request from redis", zap.Error(err))
return nil, err
}
replayed, err := ReplayRequest(routerUrl, entry.Req)
if err != nil {
logger.Error("error replaying request", zap.Error(err))
return nil, err
}
resp, err := json.Marshal(replayed)
if err != nil {
logger.Error("error marshalling replayed request response", zap.Error(err))
return nil, err
}
return resp, nil
}
func ReplayRequest(routerUrl string, request *redisCache.Request) ([]string, error) {
path := request.URL["Path"] // Includes slash prefix
payload := request.URL["Payload"]
targetUrl := fmt.Sprintf("%v%v", routerUrl, path)
var req *http.Request
var err error
client := http.DefaultClient
if request.Method == http.MethodGet {
req, err = http.NewRequest("GET", targetUrl, nil)
if err != nil {
return nil, err
}
} else {
req, err = http.NewRequest(request.Method, targetUrl, bytes.NewReader([]byte(payload)))
if err != nil {
return nil, err
}
}
req.Header.Set("X-Fission-Replayed", "true")
resp, err := client.Do(req)
if err != nil {
return nil, errors.New(fmt.Sprintf("failed to make request: %v", err))
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.New(fmt.Sprintf("failed to read response: %v", err))
}
bodyStr := string(body)
return []string{bodyStr}, nil
}
pkg/redis/spec.proto
-51
View File
@@ -1,51 +0,0 @@
/*
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<string, string> URL = 2;
map<string, string> Header = 3;
string Host = 4;
map<string, string> Form = 5;
map<string, string> 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;
}
pkg/router/functionHandler.go
-43
View File
@@ -40,7 +40,6 @@ import (
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/error/network"
executorClient "github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/redis"
"github.com/fission/fission/pkg/throttler"
"github.com/fission/fission/pkg/types"
)
@@ -54,15 +53,12 @@ type (
functionHandler struct {
logger *zap.Logger
fmap *functionServiceMap
frmap *functionRecorderMap
trmap *triggerRecorderMap
executor *executorClient.Client
function *metav1.ObjectMeta
httpTrigger *fv1.HTTPTrigger
functionMetadataMap map[string]*metav1.ObjectMeta
fnWeightDistributionList []FunctionWeightDistribution
tsRoundTripperParams *tsRoundTripperParams
recorderName string
isDebugEnv bool
svcAddrUpdateThrottler *throttler.Throttler
functionTimeoutMap map[k8stypes.UID]int
@@ -153,26 +149,6 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res
// Set forwarded host header if not exists
roundTripper.addForwardedHostHeader(req)
// TODO: Keep? --> Needed for queries encoded in URL before they're stripped by the proxy
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)
roundTripper.logger.Debug("roundtripper posted body", zap.String("body", postedBody))
req.Body = rdr2
}
}
fnMeta := roundTripper.funcHandler.function
// Metrics stuff
@@ -315,22 +291,6 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Res
functionCallCompleted(funcMetricLabels, httpMetricLabels,
overhead, time.Since(startTime), resp.ContentLength)
if len(roundTripper.funcHandler.recorderName) > 0 {
if roundTripper.funcHandler.httpTrigger != nil {
trigger := roundTripper.funcHandler.httpTrigger.Metadata.Name
redis.Record(
roundTripper.logger,
trigger,
roundTripper.funcHandler.recorderName,
req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, fnMeta.Namespace,
time.Now().UnixNano(),
)
} else {
roundTripper.logger.Error("no http trigger attached for recorder",
zap.String("recorder", roundTripper.funcHandler.recorderName))
}
}
// return response back to user
return resp, nil
} else if i >= roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1 {
@@ -435,9 +395,6 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h
fh.logger.Debug("chosen function backend's metadata", zap.Any("metadata", fh.function))
}
// set record id
setRecordRequestIDHeader(fh.recorderName, request)
// url path
setPathInfoToHeader(request)
pkg/router/functionRecorderMap.go
-66
View File
@@ -1,66 +0,0 @@
/*
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 (
"time"
"go.uber.org/zap"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/cache"
ferror "github.com/fission/fission/pkg/error"
)
type (
functionRecorderMap struct {
logger *zap.Logger
cache *cache.Cache // map[string]*fv1.Recorder
}
)
// Why do we need an expiry?
func makeFunctionRecorderMap(logger *zap.Logger, expiry time.Duration) *functionRecorderMap {
return &functionRecorderMap{
logger: logger.Named("function_recorder_map"),
cache: cache.MakeCache(expiry, 0),
}
}
func (frmap *functionRecorderMap) lookup(function string) (*fv1.Recorder, error) {
item, err := frmap.cache.Get(function)
if err != nil {
return nil, err
}
u := item.(*fv1.Recorder)
return u, nil
}
func (frmap *functionRecorderMap) assign(function string, recorder *fv1.Recorder) {
_, err := frmap.cache.Set(function, recorder)
if err != nil {
if e, ok := err.(ferror.Error); ok && e.Code == ferror.ErrorNameExists {
return
}
frmap.logger.Error("error caching recorder for function name with a different value", zap.Error(err))
}
}
func (frmap *functionRecorderMap) remove(function string) error {
return frmap.cache.Delete(function)
}
pkg/router/httpTriggers.go
+3 -63
View File
@@ -53,14 +53,13 @@ type HTTPTriggerSet struct {
functions []fv1.Function
funcStore k8sCache.Store
funcController k8sCache.Controller
recorderSet *RecorderSet
updateRouterRequestChannel chan struct{}
tsRoundTripperParams *tsRoundTripperParams
isDebugEnv bool
svcAddrUpdateThrottler *throttler.Throttler
}
func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient,
func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionClient *crd.FissionClient,
kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
httpTriggerSet := &HTTPTriggerSet{
@@ -76,9 +75,9 @@ func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, frmap *fun
isDebugEnv: isDebugEnv,
svcAddrUpdateThrottler: actionThrottler,
}
var tStore, fnStore, rStore k8sCache.Store
var tStore, fnStore k8sCache.Store
var tController, fnController k8sCache.Controller
var recorderSet *RecorderSet
if httpTriggerSet.crdClient != nil {
tStore, tController = httpTriggerSet.initTriggerController()
httpTriggerSet.triggerStore = tStore
@@ -87,8 +86,6 @@ func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, frmap *fun
httpTriggerSet.funcStore = fnStore
httpTriggerSet.funcController = fnController
}
recorderSet = MakeRecorderSet(logger, httpTriggerSet, crdClient, rStore, frmap, trmap)
httpTriggerSet.recorderSet = recorderSet
return httpTriggerSet, tStore, fnStore
}
@@ -106,11 +103,6 @@ func (ts *HTTPTriggerSet) subscribeRouter(ctx context.Context, mr *mutableRouter
go ts.syncTriggers()
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 {
ts.logger.Fatal("failed to run recorder controller")
}
}
func defaultHomeHandler(w http.ResponseWriter, r *http.Request) {
@@ -140,12 +132,6 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
continue
}
var recorderName string
recorder, err := ts.recorderSet.triggerRecorderMap.lookup(trigger.Metadata.Name)
if err == nil && recorder != nil {
recorderName = recorder.Spec.Name
}
if rr.resolveResultType != resolveResultSingleFunction && rr.resolveResultType != resolveResultMultipleFunctions {
// not implemented yet
ts.logger.Panic("resolve result type not implemented", zap.Any("type", rr.resolveResultType))
@@ -154,14 +140,11 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
fh := &functionHandler{
logger: ts.logger.Named(trigger.Metadata.Name),
fmap: ts.functionServiceMap,
frmap: ts.recorderSet.functionRecorderMap,
trmap: ts.recorderSet.triggerRecorderMap,
executor: ts.executor,
httpTrigger: &trigger,
functionMetadataMap: rr.functionMetadataMap,
fnWeightDistributionList: rr.functionWtDistributionList,
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
isDebugEnv: ts.isDebugEnv,
svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler,
functionTimeoutMap: fnTimeoutMap,
@@ -205,21 +188,12 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router
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{
logger: ts.logger.Named(m.Name),
fmap: ts.functionServiceMap,
frmap: ts.recorderSet.functionRecorderMap,
trmap: ts.recorderSet.triggerRecorderMap,
function: &m,
executor: ts.executor,
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
isDebugEnv: ts.isDebugEnv,
svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler,
functionTimeoutMap: fnTimeoutMap,
@@ -246,20 +220,11 @@ func (ts *HTTPTriggerSet) initTriggerController() (k8sCache.Store, k8sCache.Cont
trigger := obj.(*fv1.HTTPTrigger)
go createIngress(ts.logger, 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 {
if len(recorder.Spec.Triggers) == 0 {
ts.recorderSet.triggerRecorderMap.assign(trigger.Metadata.Name, recorder)
}
}
},
DeleteFunc: func(obj interface{}) {
ts.syncTriggers()
trigger := obj.(*fv1.HTTPTrigger)
go deleteIngress(ts.logger, trigger, ts.kubeClient)
go ts.recorderSet.DeleteTriggerFromRecorderMap(trigger)
},
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
oldTrigger := oldObj.(*fv1.HTTPTrigger)
@@ -285,9 +250,7 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con
ts.syncTriggers()
},
DeleteFunc: func(obj interface{}) {
function := obj.(*fv1.Function)
ts.syncTriggers()
go ts.recorderSet.DeleteFunctionFromRecorderMap(function)
},
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
oldFn := oldObj.(*fv1.Function)
@@ -317,29 +280,6 @@ 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, &fv1.Recorder{}, resyncPeriod,
k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
recorder := obj.(*fv1.Recorder)
ts.recorderSet.newRecorder(recorder)
},
DeleteFunc: func(obj interface{}) {
recorder := obj.(*fv1.Recorder)
ts.recorderSet.disableRecorder(recorder)
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldRecorder := oldObj.(*fv1.Recorder)
newRecorder := newObj.(*fv1.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())
pkg/router/recorderController.go
-135
View File
@@ -1,135 +0,0 @@
package router
import (
"go.uber.org/zap"
"k8s.io/client-go/rest"
k8sCache "k8s.io/client-go/tools/cache"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
)
type RecorderSet struct {
logger *zap.Logger
httpTriggerSet *HTTPTriggerSet
crdClient *rest.RESTClient
recStore k8sCache.Store
recController k8sCache.Controller
functionRecorderMap *functionRecorderMap
triggerRecorderMap *triggerRecorderMap
}
func MakeRecorderSet(logger *zap.Logger, httpTriggerSet *HTTPTriggerSet, crdClient *rest.RESTClient, rStore k8sCache.Store, frmap *functionRecorderMap, trmap *triggerRecorderMap) *RecorderSet {
recorderSet := &RecorderSet{
logger: logger.Named("recorder_set"),
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 *fv1.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.(*fv1.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 *fv1.Recorder) {
function := r.Spec.Function
triggers := r.Spec.Triggers
rs.logger.Info("disabling recorder",
zap.String("recorder", r.Metadata.Name),
zap.String("function", function))
// Account for function
err := rs.functionRecorderMap.remove(function)
if err != nil {
rs.logger.Error("error disabling recorder (failed to remove function from functionRecorderMap)",
zap.Error(err),
zap.String("recorder", r.Metadata.Name),
zap.String("function", function))
}
// Account for explicitly added triggers
if len(triggers) != 0 {
for _, trigger := range triggers {
err := rs.triggerRecorderMap.remove(trigger)
if err != nil {
rs.logger.Error("error disabling recorder (failed to remove triggers from triggerRecorderMap)",
zap.Error(err),
zap.String("recorder", r.Metadata.Name),
zap.String("function", function),
zap.String("trigger", trigger))
}
}
} else {
// Account for implicitly added triggers
for _, t := range rs.httpTriggerSet.triggerStore.List() {
trigger := *t.(*fv1.HTTPTrigger)
if trigger.Spec.FunctionReference.Name == function {
err := rs.triggerRecorderMap.remove(trigger.Metadata.Name)
if err != nil {
rs.logger.Error("failed to remove trigger from triggerRecorderMap",
zap.Error(err),
zap.String("recorder", r.Metadata.Name),
zap.String("function", function),
zap.String("trigger", trigger.Metadata.Name))
}
}
}
}
rs.httpTriggerSet.syncTriggers()
}
func (rs *RecorderSet) updateRecorder(old *fv1.Recorder, newer *fv1.Recorder) {
if newer.Spec.Enabled {
rs.newRecorder(newer) // TODO: Test this
} else {
rs.disableRecorder(old)
}
}
func (rs *RecorderSet) DeleteTriggerFromRecorderMap(trigger *fv1.HTTPTrigger) {
err := rs.triggerRecorderMap.remove(trigger.Metadata.Name)
if err != nil {
rs.logger.Error("failed to remove trigger from triggerRecorderMap", zap.Error(err))
}
}
func (rs *RecorderSet) DeleteFunctionFromRecorderMap(function *fv1.Function) {
err := rs.functionRecorderMap.remove(function.Metadata.Name)
if err != nil {
rs.logger.Error("failed to remove function from functionRecorderMap", zap.Error(err))
}
}
pkg/router/requesthHeader.go
-10
View File
@@ -19,10 +19,8 @@ package router
import (
"fmt"
"net/http"
"strings"
"github.com/gorilla/mux"
uuid "github.com/satori/go.uuid"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -48,11 +46,3 @@ func setPathInfoToHeader(request *http.Request) {
}
request.Header.Set("X-Fission-Full-Url", request.URL.String())
}
// setRecordRequestIDHeader set record ID to request header
func setRecordRequestIDHeader(recorderName string, request *http.Request) {
if len(recorderName) > 0 {
reqUID := "REQ" + strings.ToLower(uuid.NewV4().String())
request.Header.Set("X-Fission-ReqUID", reqUID)
}
}
pkg/router/router.go
+1 -5
View File
@@ -116,10 +116,6 @@ func Start(logger *zap.Logger, port int, executorUrl string) {
fmap := makeFunctionServiceMap(logger, time.Minute)
frmap := makeFunctionRecorderMap(logger, time.Minute)
trmap := makeTriggerRecorderMap(logger, time.Minute)
fissionClient, kubeClient, _, err := crd.MakeFissionClient()
if err != nil {
logger.Fatal("error connecting to kubernetes API", zap.Error(err))
@@ -226,7 +222,7 @@ func Start(logger *zap.Logger, port int, executorUrl string) {
zap.Bool("default", displayAccessLog))
}
triggers, _, fnStore := makeHTTPTriggerSet(logger.Named("triggerset"), fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{
triggers, _, fnStore := makeHTTPTriggerSet(logger.Named("triggerset"), fmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{
timeout: timeout,
timeoutExponent: timeoutExponent,
disableKeepAlive: disableKeepAlive,
pkg/router/router_test.go
+1 -5
View File
@@ -51,12 +51,8 @@ func TestRouter(t *testing.T) {
fmap := makeFunctionServiceMap(logger, 0)
fmap.assign(fn, testServiceUrl)
frmap := makeFunctionRecorderMap(logger, time.Minute)
trmap := makeTriggerRecorderMap(logger, time.Minute)
// HTTP trigger set with a trigger for this function
triggers, _, _ := makeHTTPTriggerSet(logger, fmap, frmap, trmap, nil, nil, nil, nil,
triggers, _, _ := makeHTTPTriggerSet(logger, fmap, nil, nil, nil, nil,
&tsRoundTripperParams{
timeout: 50 * time.Millisecond,
timeoutExponent: 2,
pkg/router/triggerRecorderMap.go
-65
View File
@@ -1,65 +0,0 @@
/*
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 (
"time"
"go.uber.org/zap"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
"github.com/fission/fission/pkg/cache"
ferror "github.com/fission/fission/pkg/error"
)
type (
triggerRecorderMap struct {
logger *zap.Logger
cache *cache.Cache // map[string]*fv1.Recorder
}
)
func makeTriggerRecorderMap(logger *zap.Logger, expiry time.Duration) *triggerRecorderMap {
return &triggerRecorderMap{
logger: logger.Named("trigger_recorder_map"),
cache: cache.MakeCache(expiry, 0),
}
}
func (trmap *triggerRecorderMap) lookup(trigger string) (*fv1.Recorder, error) {
item, err := trmap.cache.Get(trigger)
if err != nil {
return nil, err
}
u := item.(*fv1.Recorder)
return u, nil
}
func (trmap *triggerRecorderMap) assign(trigger string, recorder *fv1.Recorder) {
_, err := trmap.cache.Set(trigger, recorder)
if err != nil {
if e, ok := err.(ferror.Error); ok && e.Code == ferror.ErrorNameExists {
return
}
trmap.logger.Error("error caching recorder for function name with a different value", zap.Error(err))
}
}
func (trmap *triggerRecorderMap) remove(trigger string) error {
return trmap.cache.Delete(trigger)
}