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

Recorder CRD, Records API, Redis deployment (#818)

Browse Source
This commit is contained in:
Nafisa Shazia
2018-08-15 05:28:40 +08:00
committed by Ta-Ching Chen
parent 07ca5df8d0
commit 74a3a54543
38 changed files with 2770 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/functionHandler.go
+59
View File
@@ -17,20 +17,26 @@ limitations under the License.
package router
import (
"bytes"
"fmt"
"io/ioutil"
"log"
"net"
"net/http"
"net/http/httputil"
"net/url"
"strings"
"time"
"github.com/gorilla/mux"
"github.com/satori/go.uuid"
"github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
"github.com/fission/fission/crd"
executorClient "github.com/fission/fission/executor/client"
"github.com/fission/fission/redis"
)
type tsRoundTripperParams struct {
@@ -42,10 +48,13 @@ type tsRoundTripperParams struct {
type functionHandler struct {
fmap *functionServiceMap
frmap *functionRecorderMap
trmap *triggerRecorderMap
executor *executorClient.Client
function *metav1.ObjectMeta
httpTrigger *crd.HTTPTrigger
tsRoundTripperParams *tsRoundTripperParams
recorderName string
}
// A layer on top of http.DefaultTransport, with retries.
@@ -83,6 +92,27 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
var needExecutor, serviceUrlFromExecutor bool
var serviceUrl *url.URL
// TODO: Keep? --> Needed for queries encoded in URL before they're stripped by the proxy
var originalUrl url.URL
originalUrl = *req.URL
// Iff this request needs to be recorded, we save the body
var postedBody string
if len(roundTripper.funcHandler.recorderName) > 0 {
if req.ContentLength > 0 {
p := make([]byte, req.ContentLength)
buf, _ := ioutil.ReadAll(req.Body)
// We need two io readers because a single reader will drain the buffer, hence we keep a replacement copy
rdr1 := ioutil.NopCloser(bytes.NewBuffer(buf))
rdr2 := ioutil.NopCloser(bytes.NewBuffer(buf))
rdr1.Read(p)
postedBody = string(p)
logrus.Info(fmt.Sprintf("%v", postedBody))
req.Body = rdr2
}
}
// Metrics stuff
startTime := time.Now()
funcMetricLabels := &functionLabels{
@@ -177,6 +207,27 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
functionCallCompleted(funcMetricLabels, httpMetricLabels,
overhead, time.Since(startTime), resp.ContentLength)
// if transport.RoundTrip succeeds and it was a cached entry, then tapService
if !serviceUrlFromExecutor {
go roundTripper.funcHandler.tapService(serviceUrl)
}
trigger := ""
if roundTripper.funcHandler.httpTrigger != nil {
trigger = roundTripper.funcHandler.httpTrigger.Metadata.Name
} else {
log.Println("No trigger attached.") // Wording?
}
if len(roundTripper.funcHandler.recorderName) > 0 {
redis.Record(
trigger,
roundTripper.funcHandler.recorderName,
req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, roundTripper.funcHandler.function.Namespace,
time.Now().UnixNano(),
)
}
// return response back to user
return resp, nil
}
@@ -227,6 +278,14 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
request.Header.Add(fmt.Sprintf("X-Fission-Params-%v", k), v)
}
var reqUID string
if len(fh.recorderName) > 0 {
UID := strings.ToLower(uuid.NewV4().String())
reqUID = "REQ" + UID
request.Header.Add("X-Fission-ReqUID", reqUID)
log.Print("Record request with ReqUID: ", reqUID)
}
// system params
MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request)
router/functionRecorderMap.go
+63
View File
@@ -0,0 +1,63 @@
/*
Copyright 2018 The Fission Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package router
import (
"log"
"time"
"github.com/fission/fission"
"github.com/fission/fission/cache"
"github.com/fission/fission/crd"
)
type (
functionRecorderMap struct {
cache *cache.Cache // map[string]*crd.Recorder
}
)
// Why do we need an expiry?
func makeFunctionRecorderMap(expiry time.Duration) *functionRecorderMap {
return &functionRecorderMap{
cache: cache.MakeCache(expiry, 0),
}
}
func (frmap *functionRecorderMap) lookup(function string) (*crd.Recorder, error) {
item, err := frmap.cache.Get(function)
if err != nil {
return nil, err
}
u := item.(*crd.Recorder)
return u, nil
}
func (frmap *functionRecorderMap) assign(function string, recorder *crd.Recorder) {
err, _ := frmap.cache.Set(function, recorder)
if err != nil {
if e, ok := err.(fission.Error); ok && e.Code == fission.ErrorNameExists {
return
}
log.Printf("error caching recorder for function name with a different value: %v", err)
}
}
func (frmap *functionRecorderMap) remove(function string) error {
return frmap.cache.Delete(function)
}
router/httpTriggers.go
+69 -4
View File
@@ -49,12 +49,12 @@ type HTTPTriggerSet struct {
functions []crd.Function
funcStore k8sCache.Store
funcController k8sCache.Controller
recorderSet *RecorderSet
updateRouterRequestChannel chan struct{}
tsRoundTripperParams *tsRoundTripperParams
tsRoundTripperParams *tsRoundTripperParams
}
func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClient,
func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient,
kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
httpTriggerSet := &HTTPTriggerSet{
functionServiceMap: fmap,
@@ -66,8 +66,9 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClie
updateRouterRequestChannel: make(chan struct{}),
tsRoundTripperParams: params,
}
var tStore, fnStore k8sCache.Store
var tStore, fnStore, rStore k8sCache.Store
var tController, fnController k8sCache.Controller
var recorderSet *RecorderSet
if httpTriggerSet.crdClient != nil {
tStore, tController = httpTriggerSet.initTriggerController()
httpTriggerSet.triggerStore = tStore
@@ -76,6 +77,8 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClie
httpTriggerSet.funcStore = fnStore
httpTriggerSet.funcController = fnController
}
recorderSet = MakeRecorderSet(httpTriggerSet, crdClient, rStore, frmap, trmap)
httpTriggerSet.recorderSet = recorderSet
return httpTriggerSet, tStore, fnStore
}
@@ -92,6 +95,11 @@ func (ts *HTTPTriggerSet) subscribeRouter(ctx context.Context, mr *mutableRouter
go ts.updateRouter()
go ts.runWatcher(ctx, ts.funcController)
go ts.runWatcher(ctx, ts.triggerController)
if ts.recorderSet.recController != nil {
go ts.runWatcher(ctx, ts.recorderSet.recController)
} else {
log.Fatal("Failed to run recorder Controller")
}
}
func defaultHomeHandler(w http.ResponseWriter, r *http.Request) {
@@ -121,6 +129,14 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
continue
}
var recorderName string
recorder, err := ts.recorderSet.triggerRecorderMap.lookup(trigger.Metadata.Name)
if err == nil && recorder != nil {
recorderName = recorder.Spec.Name
}
//log.Printf("The trigger %v should be recorded: %v", trigger.Metadata.Name, doRecord)
if rr.resolveResultType != resolveResultSingleFunction {
// not implemented yet
log.Panicf("resolve result type not implemented (%v)", rr.resolveResultType)
@@ -128,10 +144,13 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
fh := &functionHandler{
fmap: ts.functionServiceMap,
frmap: ts.recorderSet.functionRecorderMap,
trmap: ts.recorderSet.triggerRecorderMap,
function: rr.functionMetadata,
executor: ts.executor,
httpTrigger: &trigger,
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
}
ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler)
@@ -158,11 +177,21 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
// triggers route into these.
for _, function := range ts.functions {
m := function.Metadata
var recorderName string
recorder, err := ts.recorderSet.functionRecorderMap.lookup(m.Name)
if err == nil && recorder != nil {
recorderName = recorder.Spec.Name
}
fh := &functionHandler{
fmap: ts.functionServiceMap,
frmap: ts.recorderSet.functionRecorderMap,
trmap: ts.recorderSet.triggerRecorderMap,
function: &m,
executor: ts.executor,
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
}
muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler)
}
@@ -186,11 +215,22 @@ func (ts *HTTPTriggerSet) initTriggerController() (k8sCache.Store, k8sCache.Cont
trigger := obj.(*crd.HTTPTrigger)
go createIngress(trigger, ts.kubeClient)
ts.syncTriggers()
// Check if this trigger's function needs to be recorded
fnRef := trigger.Spec.FunctionReference.Name
recorder, err := ts.recorderSet.functionRecorderMap.lookup(fnRef)
if err == nil && recorder != nil {
if len(recorder.Spec.Triggers) == 0 {
ts.recorderSet.triggerRecorderMap.assign(trigger.Metadata.Name, recorder)
}
} else {
log.Print("Unable to lookup function in functionRecorderMap")
}
},
DeleteFunc: func(obj interface{}) {
ts.syncTriggers()
trigger := obj.(*crd.HTTPTrigger)
go deleteIngress(trigger, ts.kubeClient)
go ts.recorderSet.DeleteTriggerFromRecorderMap(trigger)
},
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
oldTrigger := oldObj.(*crd.HTTPTrigger)
@@ -216,7 +256,9 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con
ts.syncTriggers()
},
DeleteFunc: func(obj interface{}) {
function := obj.(*crd.Function)
ts.syncTriggers()
go ts.recorderSet.DeleteFunctionFromRecorderMap(function)
},
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
oldFn := oldObj.(*crd.Function)
@@ -243,6 +285,29 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con
return store, controller
}
func (ts *HTTPTriggerSet) initRecorderController() (k8sCache.Store, k8sCache.Controller) {
resyncPeriod := 30 * time.Second
listWatch := k8sCache.NewListWatchFromClient(ts.crdClient, "recorders", metav1.NamespaceAll, fields.Everything())
store, controller := k8sCache.NewInformer(listWatch, &crd.Recorder{}, resyncPeriod,
k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
recorder := obj.(*crd.Recorder)
ts.recorderSet.newRecorder(recorder)
},
DeleteFunc: func(obj interface{}) {
recorder := obj.(*crd.Recorder)
ts.recorderSet.disableRecorder(recorder)
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldRecorder := oldObj.(*crd.Recorder)
newRecorder := newObj.(*crd.Recorder)
ts.recorderSet.updateRecorder(oldRecorder, newRecorder)
},
},
)
return store, controller
}
func (ts *HTTPTriggerSet) runWatcher(ctx context.Context, controller k8sCache.Controller) {
go func() {
controller.Run(ctx.Done())
router/recorderController.go
+118
View File
@@ -0,0 +1,118 @@
package router
import (
"github.com/fission/fission/crd"
log "github.com/sirupsen/logrus"
"k8s.io/client-go/rest"
k8sCache "k8s.io/client-go/tools/cache"
)
type RecorderSet struct {
httpTriggerSet *HTTPTriggerSet
crdClient *rest.RESTClient
recStore k8sCache.Store
recController k8sCache.Controller
functionRecorderMap *functionRecorderMap
triggerRecorderMap *triggerRecorderMap
}
func MakeRecorderSet(httpTriggerSet *HTTPTriggerSet, crdClient *rest.RESTClient, rStore k8sCache.Store, frmap *functionRecorderMap, trmap *triggerRecorderMap) *RecorderSet {
recorderSet := &RecorderSet{
httpTriggerSet: httpTriggerSet,
crdClient: crdClient,
recStore: rStore,
functionRecorderMap: frmap,
triggerRecorderMap: trmap,
}
recorderSet.recStore, recorderSet.recController = httpTriggerSet.initRecorderController()
return recorderSet
}
// All new recorders are by default enabled
func (rs *RecorderSet) newRecorder(r *crd.Recorder) {
function := r.Spec.Function
triggers := r.Spec.Triggers
// If triggers are not explicitly specified during the creation of this recorder,
// keep track of those associated with the function specified [implicitly added triggers]
needTrackByFunction := len(triggers) == 0
rs.functionRecorderMap.assign(function, r)
if needTrackByFunction {
for _, t := range rs.httpTriggerSet.triggerStore.List() {
trigger := *t.(*crd.HTTPTrigger)
if trigger.Spec.FunctionReference.Name == function {
rs.triggerRecorderMap.assign(trigger.Metadata.Name, r)
}
}
} else {
for _, trigger := range triggers {
rs.triggerRecorderMap.assign(trigger, r)
}
}
rs.httpTriggerSet.syncTriggers()
}
// TODO: Delete or disable?
func (rs *RecorderSet) disableRecorder(r *crd.Recorder) {
function := r.Spec.Function
triggers := r.Spec.Triggers
log.Info("Disabling recorder ", r.Metadata.Name)
// Account for function
err := rs.functionRecorderMap.remove(function)
if err != nil {
log.Error("Error disabling recorder (failed to remove function from functionRecorderMap): ", err)
}
// Account for explicitly added triggers
if len(triggers) != 0 {
for _, trigger := range triggers {
err := rs.triggerRecorderMap.remove(trigger)
if err != nil {
log.Error("Error disabling recorder (failed to remove triggers from triggerRecorderMap): ", err)
}
}
} else {
// Account for implicitly added triggers
for _, t := range rs.httpTriggerSet.triggerStore.List() {
trigger := *t.(*crd.HTTPTrigger)
if trigger.Spec.FunctionReference.Name == function {
err := rs.triggerRecorderMap.remove(trigger.Metadata.Name)
if err != nil {
log.Error("Failed to remove trigger from triggerRecorderMap: ", err)
}
}
}
}
rs.httpTriggerSet.syncTriggers()
}
func (rs *RecorderSet) updateRecorder(old *crd.Recorder, newer *crd.Recorder) {
if newer.Spec.Enabled == true {
rs.newRecorder(newer) // TODO: Test this
} else {
rs.disableRecorder(old)
}
}
func (rs *RecorderSet) DeleteTriggerFromRecorderMap(trigger *crd.HTTPTrigger) {
err := rs.triggerRecorderMap.remove(trigger.Metadata.Name)
if err != nil {
log.Error("Failed to remove trigger from triggerRecorderMap: ", err)
}
}
func (rs *RecorderSet) DeleteFunctionFromRecorderMap(function *crd.Function) {
err := rs.functionRecorderMap.remove(function.Metadata.Name)
if err != nil {
log.Error("Failed to remove function from functionRecorderMap: ", err)
}
}
router/router.go
+6 -1
View File
@@ -86,6 +86,10 @@ func Start(port int, executorUrl string) {
fmap := makeFunctionServiceMap(time.Minute)
frmap := makeFunctionRecorderMap(time.Minute)
trmap := makeTriggerRecorderMap(time.Minute)
fissionClient, kubeClient, _, err := crd.MakeFissionClient()
if err != nil {
log.Fatalf("Error connecting to kubernetes API: %v", err)
@@ -120,13 +124,14 @@ func Start(port int, executorUrl string) {
log.Fatalf("Failed to parse max retry times: %v", err)
}
triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, kubeClient, executor, restClient,
triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient,
&tsRoundTripperParams{
timeout: timeout,
timeoutExponent: timeoutExponent,
keepAlive: keepAlive,
maxRetries: maxRetries,
})
resolver := makeFunctionReferenceResolver(fnStore)
go serveMetric()
router/router_test.go
+5 -1
View File
@@ -58,8 +58,12 @@ func TestRouter(t *testing.T) {
}
frr.refCache.Set(nfr, rr)
frmap := makeFunctionRecorderMap(time.Minute)
trmap := makeTriggerRecorderMap(time.Minute)
// HTTP trigger set with a trigger for this function
triggers, _, _ := makeHTTPTriggerSet(fmap, nil, nil, nil, nil,
triggers, _, _ := makeHTTPTriggerSet(fmap, frmap, trmap, nil, nil, nil, nil,
&tsRoundTripperParams{
timeout: 50 * time.Millisecond,
timeoutExponent: 2,
router/triggerRecorderMap.go
+62
View File
@@ -0,0 +1,62 @@
/*
Copyright 2018 The Fission Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package router
import (
"log"
"time"
"github.com/fission/fission"
"github.com/fission/fission/cache"
"github.com/fission/fission/crd"
)
type (
triggerRecorderMap struct {
cache *cache.Cache // map[string]*crd.Recorder
}
)
func makeTriggerRecorderMap(expiry time.Duration) *triggerRecorderMap {
return &triggerRecorderMap{
cache: cache.MakeCache(expiry, 0),
}
}
func (trmap *triggerRecorderMap) lookup(trigger string) (*crd.Recorder, error) {
item, err := trmap.cache.Get(trigger)
if err != nil {
return nil, err
}
u := item.(*crd.Recorder)
return u, nil
}
func (trmap *triggerRecorderMap) assign(trigger string, recorder *crd.Recorder) {
err, _ := trmap.cache.Set(trigger, recorder)
if err != nil {
if e, ok := err.(fission.Error); ok && e.Code == fission.ErrorNameExists {
return
}
log.Printf("error caching recorder for function name with a different value: %v", err)
}
}
func (trmap *triggerRecorderMap) remove(trigger string) error {
return trmap.cache.Delete(trigger)
}