diff --git a/controller/api.go b/controller/api.go index 875b0bc3..c6342277 100644 --- a/controller/api.go +++ b/controller/api.go @@ -32,6 +32,7 @@ import ( type API struct { FunctionStore HTTPTriggerStore + TimeTriggerStore EnvironmentStore WatchStore } @@ -40,6 +41,7 @@ func MakeAPI(rs *ResourceStore) *API { api := &API{ FunctionStore: FunctionStore{ResourceStore: *rs}, HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs}, + TimeTriggerStore: TimeTriggerStore{ResourceStore: *rs}, EnvironmentStore: EnvironmentStore{ResourceStore: *rs}, WatchStore: WatchStore{ResourceStore: *rs}, } @@ -95,6 +97,12 @@ func (api *API) Serve(port int) { r.HandleFunc("/v1/watches/{watch}", api.WatchApiUpdate).Methods("PUT") r.HandleFunc("/v1/watches/{watch}", api.WatchApiDelete).Methods("DELETE") + r.HandleFunc("/v1/triggers/time", api.TimeTriggerApiList).Methods("GET") + r.HandleFunc("/v1/triggers/time", api.TimeTriggerApiCreate).Methods("POST") + r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiGet).Methods("GET") + r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiUpdate).Methods("PUT") + r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiDelete).Methods("DELETE") + address := fmt.Sprintf(":%v", port) log.WithFields(log.Fields{"port": port}).Info("Server started") diff --git a/controller/api_test.go b/controller/api_test.go index f6f523ac..40476a3e 100644 --- a/controller/api_test.go +++ b/controller/api_test.go @@ -50,6 +50,13 @@ func assertNotFoundFails(err error, name string) { assert(fe.Code == fission.ErrorNotFound, "error must be a not found error") } +func assertCronSpecFails(err error) { + assert(err != nil, "using an invalid cron spec must fail") + fe, ok := err.(fission.Error) + assert(ok, "error must be a fission Error") + assert(fe.Code == fission.ErrorInvalidArgument, "error must be a invalid argument error") +} + func TestFunctionApi(t *testing.T) { log.SetFormatter(&log.TextFormatter{DisableColors: true}) @@ -331,6 +338,53 @@ func TestWatchApi(t *testing.T) { assert(len(ws) == 2, "created two envs, but didn't find them") } +func TestTimeTriggerApi(t *testing.T) { + testTrigger := &fission.TimeTrigger{ + Metadata: fission.Metadata{ + Name: "xxx", + Uid: "yyy", + }, + Cron: "0 30 * * * *", + Function: fission.Metadata{ + Name: "foo", + Uid: "", + }, + } + _, err := g.client.TimeTriggerGet(&fission.Metadata{Name: "foo"}) + assertNotFoundFails(err, "trigger") + + m, err := g.client.TimeTriggerCreate(testTrigger) + panicIf(err) + defer g.client.TimeTriggerDelete(m) + + _, err = g.client.TimeTriggerCreate(testTrigger) + assertNameReuseFails(err, "trigger") + + tr, err := g.client.TimeTriggerGet(m) + panicIf(err) + testTrigger.Metadata.Uid = m.Uid + assert(*testTrigger == *tr, "trigger should match after reading") + + testTrigger.Cron = "@hourly" + m2, err := g.client.TimeTriggerUpdate(testTrigger) + panicIf(err) + + m.Uid = m2.Uid + tr, err = g.client.TimeTriggerGet(m) + panicIf(err) + testTrigger.Metadata.Uid = m.Uid + assert(*testTrigger == *tr, "trigger should match after reading") + + testTrigger.Metadata.Name = "yyy" + testTrigger.Cron = "Not valid cron spec" + m, err = g.client.TimeTriggerCreate(testTrigger) + assertCronSpecFails(err) + + ts, err := g.client.TimeTriggerList() + panicIf(err) + assert(len(ts) == 1, "created one trigger, but didn't find it") +} + func TestMain(m *testing.M) { flag.Parse() @@ -342,6 +396,7 @@ func TestMain(m *testing.M) { ks.Delete(context.Background(), "Function", &etcdClient.DeleteOptions{Recursive: true}) ks.Delete(context.Background(), "HTTPTrigger", &etcdClient.DeleteOptions{Recursive: true}) + ks.Delete(context.Background(), "TimeTrigger", &etcdClient.DeleteOptions{Recursive: true}) ks.Delete(context.Background(), "Environment", &etcdClient.DeleteOptions{Recursive: true}) ks.Delete(context.Background(), "Watch", &etcdClient.DeleteOptions{Recursive: true}) diff --git a/controller/client/client.go b/controller/client/client.go index eac75e1a..ba880f15 100644 --- a/controller/client/client.go +++ b/controller/client/client.go @@ -528,3 +528,110 @@ func (c *Client) WatchList() ([]fission.Watch, error) { return watches, err } + +func (c *Client) TimeTriggerCreate(t *fission.TimeTrigger) (*fission.Metadata, error) { + reqbody, err := json.Marshal(t) + if err != nil { + return nil, err + } + + resp, err := http.Post(c.url("triggers/time"), "application/json", bytes.NewReader(reqbody)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleCreateResponse(resp) + if err != nil { + return nil, err + } + + var m fission.Metadata + err = json.Unmarshal(body, &m) + if err != nil { + return nil, err + } + + return &m, nil +} + +func (c *Client) TimeTriggerGet(m *fission.Metadata) (*fission.TimeTrigger, error) { + relativeUrl := fmt.Sprintf("triggers/time/%v", m.Name) + if len(m.Uid) > 0 { + relativeUrl += fmt.Sprintf("?uid=%v", m.Uid) + } + + resp, err := http.Get(c.url(relativeUrl)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + var t fission.TimeTrigger + err = json.Unmarshal(body, &t) + if err != nil { + return nil, err + } + + return &t, nil +} + +func (c *Client) TimeTriggerUpdate(t *fission.TimeTrigger) (*fission.Metadata, error) { + reqbody, err := json.Marshal(t) + if err != nil { + return nil, err + } + relativeUrl := fmt.Sprintf("triggers/time/%v", t.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 fission.Metadata + err = json.Unmarshal(body, &m) + if err != nil { + return nil, err + } + return &m, nil +} + +func (c *Client) TimeTriggerDelete(m *fission.Metadata) error { + relativeUrl := fmt.Sprintf("triggers/time/%v", m.Name) + if len(m.Uid) > 0 { + relativeUrl += fmt.Sprintf("?uid=%v", m.Uid) + } + err := c.delete(relativeUrl) + return err +} + +func (c *Client) TimeTriggerList() ([]fission.TimeTrigger, error) { + resp, err := http.Get(c.url("triggers/time")) + if err != nil { + return nil, err + } + + body, err := c.handleResponse(resp) + if err != nil { + return nil, err + } + + triggers := make([]fission.TimeTrigger, 0) + err = json.Unmarshal(body, &triggers) + if err != nil { + return nil, err + } + + return triggers, nil +} diff --git a/controller/timeTriggerApi.go b/controller/timeTriggerApi.go new file mode 100644 index 00000000..501f5681 --- /dev/null +++ b/controller/timeTriggerApi.go @@ -0,0 +1,181 @@ +/* +Copyright 2017 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" + + log "github.com/Sirupsen/logrus" + "github.com/gorilla/mux" + "github.com/robfig/cron" + + "github.com/fission/fission" +) + +func (api *API) TimeTriggerApiList(w http.ResponseWriter, r *http.Request) { + triggers, err := api.TimeTriggerStore.List() + if err != nil { + api.respondWithError(w, err) + return + } + + resp, err := json.Marshal(triggers) + if err != nil { + api.respondWithError(w, err) + return + } + + api.respondWithSuccess(w, resp) +} + +func (api *API) TimeTriggerApiCreate(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + api.respondWithError(w, err) + } + + var t fission.TimeTrigger + err = json.Unmarshal(body, &t) + if err != nil { + api.respondWithError(w, err) + return + } + + triggers, err := api.TimeTriggerStore.List() + if err != nil { + api.respondWithError(w, err) + return + } + for _, trigger := range triggers { + if trigger.Name == t.Name { + err = fission.MakeError(fission.ErrorNameExists, + "TimeTrigger with same name already exists") + api.respondWithError(w, err) + return + } + } + + _, err = cron.Parse(t.Cron) + if err != nil { + err = fission.MakeError(fission.ErrorInvalidArgument, "TimeTrigger cron spec is not valid") + api.respondWithError(w, err) + return + } + + uid, err := api.TimeTriggerStore.Create(&t) + if err != nil { + api.respondWithError(w, err) + return + } + + m := &fission.Metadata{Name: t.Metadata.Name, Uid: uid} + resp, err := json.Marshal(m) + if err != nil { + api.respondWithError(w, err) + return + } + + w.WriteHeader(http.StatusCreated) + api.respondWithSuccess(w, resp) +} + +func (api *API) TimeTriggerApiGet(w http.ResponseWriter, r *http.Request) { + var m fission.Metadata + + vars := mux.Vars(r) + m.Name = vars["timeTrigger"] + m.Uid = r.FormValue("uid") // empty if uid is absent + + t, err := api.TimeTriggerStore.Get(&m) + if err != nil { + api.respondWithError(w, err) + return + } + + resp, err := json.Marshal(t) + if err != nil { + api.respondWithError(w, err) + return + } + + api.respondWithSuccess(w, resp) +} + +func (api *API) TimeTriggerApiUpdate(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + name := vars["timeTrigger"] + + body, err := ioutil.ReadAll(r.Body) + if err != nil { + api.respondWithError(w, err) + } + + var t fission.TimeTrigger + err = json.Unmarshal(body, &t) + if err != nil { + api.respondWithError(w, err) + return + } + + if name != t.Metadata.Name { + err = fission.MakeError(fission.ErrorInvalidArgument, "TimeTrigger name doesn't match URL") + api.respondWithError(w, err) + return + } + + _, err = cron.Parse(t.Cron) + if err != nil { + err = fission.MakeError(fission.ErrorInvalidArgument, "TimeTrigger cron spec is not valid") + api.respondWithError(w, err) + return + } + + uid, err := api.TimeTriggerStore.Update(&t) + if err != nil { + api.respondWithError(w, err) + return + } + + m := &fission.Metadata{Name: t.Metadata.Name, Uid: uid} + resp, err := json.Marshal(m) + if err != nil { + api.respondWithError(w, err) + return + } + api.respondWithSuccess(w, resp) +} + +func (api *API) TimeTriggerApiDelete(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + var m fission.Metadata + m.Name = vars["timeTrigger"] + + m.Uid = r.FormValue("uid") // empty if uid is absent + if len(m.Uid) == 0 { + log.WithFields(log.Fields{"timeTrigger": m.Name}).Info("Deleting all versions") + } + + err := api.TimeTriggerStore.Delete(m) + if err != nil { + api.respondWithError(w, err) + return + } + + api.respondWithSuccess(w, []byte("")) +} diff --git a/controller/timeTriggerStore.go b/controller/timeTriggerStore.go new file mode 100644 index 00000000..b3a44c0c --- /dev/null +++ b/controller/timeTriggerStore.go @@ -0,0 +1,79 @@ +/* +Copyrigtt 2017 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + tttp://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "github.com/satori/go.uuid" + + "github.com/fission/fission" +) + +type TimeTriggerStore struct { + ResourceStore +} + +func (tts *TimeTriggerStore) Create(tt *fission.TimeTrigger) (string, error) { + tt.Metadata.Uid = uuid.NewV4().String() + return tt.Metadata.Uid, tts.ResourceStore.create(tt) +} + +func (tts *TimeTriggerStore) Get(m *fission.Metadata) (*fission.TimeTrigger, error) { + var tt fission.TimeTrigger + err := tts.ResourceStore.read(m.Name, &tt) + if err != nil { + return nil, err + } + return &tt, nil +} + +func (tts *TimeTriggerStore) Update(tt *fission.TimeTrigger) (string, error) { + tt.Metadata.Uid = uuid.NewV4().String() + return tt.Metadata.Uid, tts.ResourceStore.update(tt) +} + +func (tts *TimeTriggerStore) Delete(m fission.Metadata) error { + typeName, err := getTypeName(fission.TimeTrigger{}) + if err != nil { + return err + } + return tts.ResourceStore.delete(typeName, m.Name) +} + +func (tts *TimeTriggerStore) List() ([]fission.TimeTrigger, error) { + typeName, err := getTypeName(fission.TimeTrigger{}) + if err != nil { + return nil, err + } + + bufs, err := tts.ResourceStore.getAll(typeName) + if err != nil { + return nil, err + } + + triggers := make([]fission.TimeTrigger, 0, len(bufs)) + js := JsonSerializer{} + for _, buf := range bufs { + var tt fission.TimeTrigger + err = js.deserialize([]byte(buf), &tt) + if err != nil { + return nil, err + } + triggers = append(triggers, tt) + } + + return triggers, nil +} diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 010615b1..56282cde 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -10,6 +10,7 @@ import ( "github.com/fission/fission/logger" "github.com/fission/fission/poolmgr" "github.com/fission/fission/router" + "github.com/fission/fission/timer" ) func runController(port int, etcdUrl string, filepath string) { @@ -53,6 +54,13 @@ func runLogger() { log.Fatalf("Error: Logger exited.") } +func runTimer(controllerUrl, routerUrl string) { + err := timer.Start(controllerUrl, routerUrl) + if err != nil { + log.Fatalf("Error starting timer: %v", err) + } +} + func getPort(portArg interface{}) int { portArgStr := portArg.(string) port, err := strconv.Atoi(portArgStr) @@ -87,6 +95,7 @@ Usage: fission-bundle --poolmgrPort= [--controllerUrl= --namespace=] fission-bundle --kubewatcher [--controllerUrl= --routerUrl=] fission-bundle --logger + fission-bundle --timer [--controllerUrl= --routerUrl=] Options: --controllerPort= Port that the controller should listen on. --routerPort= Port that the router should listen on. @@ -99,6 +108,7 @@ Options: --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. --kubewatcher Start Kubernetes events watcher. --logger Start logger. + --timer Start Timer. ` arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false) if err != nil { @@ -135,5 +145,9 @@ Options: runLogger() } + if arguments["--timer"] == true { + runTimer(controllerUrl, routerUrl) + } + select {} } diff --git a/fission.yaml b/fission.yaml index cc744bf5..235f9479 100644 --- a/fission.yaml +++ b/fission.yaml @@ -104,6 +104,25 @@ spec: command: ["/fission-bundle"] args: ["--kubewatcher"] +--- +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: timer + namespace: fission +spec: + replicas: 1 + template: + metadata: + labels: + svc: timer + spec: + containers: + - name: timer + image: fission/fission-bundle:alpha20170124 + command: ["/fission-bundle"] + args: ["--timer"] + --- apiVersion: v1 kind: Service diff --git a/fission/main.go b/fission/main.go index f74efa45..9fb0fa19 100644 --- a/fission/main.go +++ b/fission/main.go @@ -72,6 +72,19 @@ func main() { {Name: "list", Usage: "List HTTP triggers", Flags: []cli.Flag{}, Action: htList}, } + // timetriggers + ttNameFlag := cli.StringFlag{Name: "name", Usage: "Time Trigger name"} + ttCronFlag := cli.StringFlag{Name: "cron", Usage: "Time Trigger cron spec ('0 30 * * *', '@every 5m', '@hourly')"} + ttFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"} + ttFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"} + ttSubcommands := []cli.Command{ + {Name: "create", Aliases: []string{"add"}, Usage: "Create Time trigger", Flags: []cli.Flag{ttNameFlag, ttFnNameFlag, ttFnUidFlag, ttCronFlag}, Action: ttCreate}, + {Name: "get", Usage: "Get Time trigger", Flags: []cli.Flag{}, Action: ttGet}, + {Name: "update", Usage: "Update Time trigger", Flags: []cli.Flag{ttNameFlag, ttCronFlag}, Action: ttUpdate}, + {Name: "delete", Usage: "Delete Time trigger", Flags: []cli.Flag{ttNameFlag}, Action: ttDelete}, + {Name: "list", Usage: "List Time triggers", Flags: []cli.Flag{}, Action: ttList}, + } + // environments envNameFlag := cli.StringFlag{Name: "name", Usage: "Environment name"} envImageFlag := cli.StringFlag{Name: "image", Usage: "Environment image URL"} @@ -101,6 +114,7 @@ func main() { app.Commands = []cli.Command{ {Name: "function", Aliases: []string{"fn"}, Usage: "Create, update and manage functions", Subcommands: fnSubcommands}, {Name: "httptrigger", Aliases: []string{"ht", "route"}, Usage: "Manage HTTP triggers (routes) for functions", Subcommands: htSubcommands}, + {Name: "timetrigger", Aliases: []string{"tt", "timer"}, Usage: "Manage Time triggers (timers) for functions", Subcommands: ttSubcommands}, {Name: "environment", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands}, {Name: "watch", Aliases: []string{"w"}, Usage: "Manage watches", Subcommands: wSubCommands}, diff --git a/fission/timetrigger.go b/fission/timetrigger.go new file mode 100644 index 00000000..a7776851 --- /dev/null +++ b/fission/timetrigger.go @@ -0,0 +1,122 @@ +/* +Copyrigtt 2017 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + tttp://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package main + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/satori/go.uuid" + "github.com/urfave/cli" + + "github.com/fission/fission" +) + +func ttCreate(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + name := c.String("name") + if len(name) == 0 { + name = uuid.NewV4().String() + } + fnName := c.String("function") + if len(fnName) == 0 { + fatal("Need a function name to create a trigger, use --function") + } + fnUid := c.String("uid") + cron := c.String("cron") + if len(cron) == 0 { + fatal("Need a cron spec like '0 30 * * *', '@every 1h30m', '@hourly', use --cron") + } + + tt := &fission.TimeTrigger{ + Metadata: fission.Metadata{ + Name: name, + }, + Cron: cron, + Function: fission.Metadata{ + Name: fnName, + Uid: fnUid, + }, + } + + _, err := client.TimeTriggerCreate(tt) + checkErr(err, "create Time trigger") + + fmt.Printf("trigger '%v' created\n", name) + return err +} + +func ttGet(c *cli.Context) error { + return nil +} + +func ttUpdate(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + ttName := c.String("name") + if len(ttName) == 0 { + fatal("Need name of trigger, use --name") + } + + tt, err := client.TimeTriggerGet(&fission.Metadata{Name: ttName}) + checkErr(err, "get Time trigger") + + newCron := c.String("cron") + if len(newCron) != 0 { + tt.Cron = newCron + } + + _, err = client.TimeTriggerUpdate(tt) + checkErr(err, "update Time trigger") + + fmt.Printf("trigger '%v' updated\n", ttName) + return nil +} + +func ttDelete(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + ttName := c.String("name") + if len(ttName) == 0 { + fatal("Need name of trigger to delete, use --name") + } + + err := client.TimeTriggerDelete(&fission.Metadata{Name: ttName}) + checkErr(err, "delete trigger") + + fmt.Printf("trigger '%v' deleted\n", ttName) + return nil +} + +func ttList(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + tts, err := client.TimeTriggerList() + checkErr(err, "list Time triggers") + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + + fmt.Fprintf(w, "%v\t%v\t%v\t%v\n", + "NAME", "CRON", "FUNCTION_NAME", "FUNCTION_UID") + for _, tt := range tts { + fmt.Fprintf(w, "%v\t%v\t%v\t%v\n", + tt.Metadata.Name, tt.Cron, tt.Function.Name, tt.Function.Uid) + } + w.Flush() + + return nil +} diff --git a/glide.lock b/glide.lock index b88c5d10..6464ed29 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 6b4c050c1cd5a7cbb99141348020c882d088b30a9fe9cf92b78a7f706e59da4c -updated: 2017-02-22T23:15:19.862789385+08:00 +hash: 51b5eda27974aac228f42002cb0a9f89bf2f79d36a574a3694c142c2f83e4042 +updated: 2017-03-24T11:01:00.667996869+08:00 imports: - name: github.com/blang/semver version: 60ec3488bfea7cca02b021d106d9911120d25fe9 @@ -94,10 +94,12 @@ imports: version: 8a290539e2e8629dbc4e6bad948158f790ec31f4 - name: github.com/PuerkitoBio/urlesc version: 5bd2802263f21d8788851d5305584c82a5c75d7e +- name: github.com/robfig/cron + version: df38d32658d8788cd446ba74db4bb5375c4b0cb3 - name: github.com/satori/go.uuid version: 879c5887cd475cd7864858769793b2ceb0d44feb - name: github.com/Sirupsen/logrus - version: c078b1e43f58d563c74cebe63c85789e76ddb627 + version: ba1b36c82c5e05c4f912a88eab0dcd91a171688f - name: github.com/spf13/pflag version: 08b1a584251b5b62f458943640fc8ebd4d50aaa5 - name: github.com/ugorji/go diff --git a/glide.yaml b/glide.yaml index 80463c84..a06622fc 100644 --- a/glide.yaml +++ b/glide.yaml @@ -33,3 +33,4 @@ import: version: v1.2.0 subpackages: - client/v2 +- package: github.com/robfig/cron diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index 62537a1e..c7198257 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -35,6 +35,8 @@ import ( "k8s.io/client-go/1.5/pkg/watch" "github.com/fission/fission" + "github.com/fission/fission/publisher" + "reflect" ) type requestType int @@ -48,7 +50,8 @@ type ( watches map[string]watchSubscription kubernetesClient *kubernetes.Clientset requestChannel chan *kubeWatcherRequest - publisher Publisher + publisher publisher.Publisher + routerUrl string } watchSubscription struct { @@ -57,7 +60,7 @@ type ( lastResourceVersion string stopped *int32 kubernetesClient *kubernetes.Clientset - publisher Publisher + publisher publisher.Publisher } kubeWatcherRequest struct { @@ -70,7 +73,7 @@ type ( } ) -func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, publisher Publisher) *KubeWatcher { +func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, publisher publisher.Publisher) *KubeWatcher { kw := &KubeWatcher{ watches: make(map[string]watchSubscription), kubernetesClient: kubernetesClient, @@ -204,7 +207,7 @@ func (kw *KubeWatcher) removeWatch(w *fission.Watch) error { // return nil // } -func MakeWatchSubscription(w *fission.Watch, kubeClient *kubernetes.Clientset, publisher Publisher) (*watchSubscription, error) { +func MakeWatchSubscription(w *fission.Watch, kubeClient *kubernetes.Clientset, publisher publisher.Publisher) (*watchSubscription, error) { var stopped int32 = 0 ws := &watchSubscription{ Watch: *w, @@ -277,7 +280,22 @@ func (ws *watchSubscription) eventDispatchLoop() { ws.lastResourceVersion = rv } - ws.publisher.Publish(ev, ws.Watch.Target) + // Serialize the object + var buf bytes.Buffer + err = printKubernetesObject(ev.Object, &buf) + if err != nil { + log.Printf("Failed to serialize object: %v", err) + // TODO send a POST request indicating error + } + + // Event and object type aren't in the serialized object + headers := map[string]string{ + "Content-Type": "application/json", + "X-Kubernetes-Event-Type": string(ev.Type), + "X-Kubernetes-Object-Type": reflect.TypeOf(ev.Object).Elem().Name(), + } + // Event and object type aren't in the serialized object + ws.publisher.Publish(buf.String(), headers, ws.Watch.Target) } if atomic.LoadInt32(ws.stopped) == 0 { err := ws.restartWatch() diff --git a/kubewatcher/main.go b/kubewatcher/main.go index de19772c..28118fc5 100644 --- a/kubewatcher/main.go +++ b/kubewatcher/main.go @@ -23,6 +23,7 @@ import ( "k8s.io/client-go/1.5/rest" "github.com/fission/fission/controller/client" + "github.com/fission/fission/publisher" ) // Get a kubernetes client using the pod's service account. @@ -49,11 +50,10 @@ func Start(controllerUrl string, routerUrl string) error { if err != nil { return err } - poster := MakeWebhookPublisher(routerUrl) + poster := publisher.MakeWebhookPublisher(routerUrl) kubeWatch := MakeKubeWatcher(kubeClient, poster) - client := client.MakeClient(controllerUrl) - MakeWatchSync(client, kubeWatch) + MakeWatchSync(client.MakeClient(controllerUrl), kubeWatch) return nil } diff --git a/kubewatcher/publisher.go b/publisher/publisher.go similarity index 80% rename from kubewatcher/publisher.go rename to publisher/publisher.go index 7d07e24f..7c2a77c7 100644 --- a/kubewatcher/publisher.go +++ b/publisher/publisher.go @@ -14,17 +14,13 @@ See the License for the specific language governing permissions and limitations under the License. */ -package kubewatcher - -import ( - "k8s.io/client-go/1.5/pkg/watch" -) +package publisher type ( Publisher interface { - // Publish an event to a "target". Target's meaning depends on the + // Publish an request to a "target". Target's meaning depends on the // publisher: it's a URL in the case of a webhook publisher, or a queue // name in a queue-based publisher such as NATS. - Publish(event watch.Event, target string) + Publish(body string, headers map[string]string, target string) } ) diff --git a/kubewatcher/webhookPublisher.go b/publisher/webhookPublisher.go similarity index 71% rename from kubewatcher/webhookPublisher.go rename to publisher/webhookPublisher.go index b37687a4..4cda32f5 100644 --- a/kubewatcher/webhookPublisher.go +++ b/publisher/webhookPublisher.go @@ -1,5 +1,5 @@ /* -Copyright 2016 The Fission Authors. +Copyright 2017 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. @@ -14,18 +14,15 @@ See the License for the specific language governing permissions and limitations under the License. */ -package kubewatcher +package publisher import ( "bytes" "io/ioutil" "log" "net/http" - "reflect" "strings" "time" - - "k8s.io/client-go/1.5/pkg/watch" ) type ( @@ -39,8 +36,9 @@ type ( baseUrl string } publishRequest struct { - url string - watchEvent watch.Event + body string + headers map[string]string + target string retries int retryDelay time.Duration } @@ -58,10 +56,11 @@ func MakeWebhookPublisher(baseUrl string) *WebhookPublisher { return p } -func (p *WebhookPublisher) Publish(watchEvent watch.Event, url string) { +func (p *WebhookPublisher) Publish(body string, headers map[string]string, target string) { p.requestChannel <- &publishRequest{ - watchEvent: watchEvent, - url: url, + body: body, + headers: headers, + target: target, retries: p.maxRetries, retryDelay: p.retryDelay, } @@ -75,31 +74,18 @@ func (p *WebhookPublisher) svc() { } func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { - - url := p.baseUrl + "/" + strings.TrimPrefix(r.url, "/") + url := p.baseUrl + "/" + strings.TrimPrefix(r.target, "/") log.Printf("Making HTTP request to %v", url) - // Serialize the object var buf bytes.Buffer - err := printKubernetesObject(r.watchEvent.Object, &buf) - if err != nil { - log.Printf("Failed to serialize object: %v", err) - // TODO send a POST request indicating error - } + buf.WriteString(r.body) // Create request req, err := http.NewRequest("POST", url, &buf) - if err != nil { - log.Printf("Failed to create request to %v", url) - // can't do anything more, drop the event. - return + for k, v := range r.headers { + req.Header.Add(k, v) } - // Event and object type aren't in the serialized object - req.Header.Add("Content-Type", "application/json") - req.Header.Add("X-Kubernetes-Event-Type", string(r.watchEvent.Type)) - req.Header.Add("X-Kubernetes-Object-Type", reflect.TypeOf(r.watchEvent.Object).Elem().Name()) - // Make the request resp, err := http.DefaultClient.Do(req) diff --git a/resource.go b/resource.go index 5695669f..8f41bf32 100644 --- a/resource.go +++ b/resource.go @@ -28,6 +28,10 @@ func (ht HTTPTrigger) Key() string { return ht.Metadata.Name } +func (tt TimeTrigger) Key() string { + return tt.Metadata.Name +} + func (w Watch) Key() string { return w.Metadata.Name } diff --git a/timer/main.go b/timer/main.go new file mode 100644 index 00000000..aff393e8 --- /dev/null +++ b/timer/main.go @@ -0,0 +1,30 @@ +/* +Copyright 2017 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 timer + +import ( + controllerClient "github.com/fission/fission/controller/client" + "github.com/fission/fission/publisher" +) + +func Start(controllerUrl string, routerUrl string) error { + controller := controllerClient.MakeClient(controllerUrl) + poster := publisher.MakeWebhookPublisher(routerUrl) + MakeTimerSync(controller, MakeTimer(poster)) + + return nil +} diff --git a/timer/timer.go b/timer/timer.go new file mode 100644 index 00000000..03a73b55 --- /dev/null +++ b/timer/timer.go @@ -0,0 +1,143 @@ +/* +Copyright 2017 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 timer + +import ( + "github.com/robfig/cron" + + "log" + + "github.com/fission/fission" + "github.com/fission/fission/publisher" +) + +type requestType int + +const ( + SYNC requestType = iota +) + +type ( + Timer struct { + triggers map[string]*timerTriggerWithCron + requestChannel chan *timerRequest + publisher *publisher.Publisher + } + + timerRequest struct { + requestType + triggers []fission.TimeTrigger + responseChannel chan *timerResponse + } + timerResponse struct { + error + } + timerTriggerWithCron struct { + trigger fission.TimeTrigger + cron *cron.Cron + } +) + +func MakeTimer(publisher publisher.Publisher) *Timer { + timer := &Timer{ + triggers: make(map[string]*timerTriggerWithCron), + requestChannel: make(chan *timerRequest), + publisher: &publisher, + } + go timer.svc() + return timer +} + +func (timer *Timer) Sync(triggers []fission.TimeTrigger) error { + req := &timerRequest{ + requestType: SYNC, + triggers: triggers, + responseChannel: make(chan *timerResponse), + } + timer.requestChannel <- req + resp := <-req.responseChannel + return resp.error +} + +func (timer *Timer) svc() { + for { + req := <-timer.requestChannel + switch req.requestType { + case SYNC: + err := timer.syncCron(req.triggers) + req.responseChannel <- &timerResponse{error: err} + } + } +} + +func (timer *Timer) syncCron(triggers []fission.TimeTrigger) error { + for _, t := range triggers { + if item, ok := timer.triggers[t.Name]; ok { + // the item exists, update the item if needed + if item.trigger.Uid == t.Uid { + continue + } + + // update cron if the cron spec changed + if item.trigger.Cron != t.Cron { + // if there is an cron running, stop it + if item.cron != nil { + item.cron.Stop() + } + item.cron = timer.newCron(t) + } + + item.trigger = t + } else { + timer.triggers[t.Name] = &timerTriggerWithCron{ + trigger: t, + cron: timer.newCron(t), + } + } + } + + for k, v := range timer.triggers { + found := false + for _, t := range triggers { + if t.Name == k { + found = true + break + } + } + if !found { + if v.cron != nil { + v.cron.Stop() + } + delete(timer.triggers, k) + } + } + + return nil +} + +func (timer *Timer) newCron(t fission.TimeTrigger) *cron.Cron { + c := cron.New() + c.AddFunc(t.Cron, func() { + headers := map[string]string{ + "X-Fission-Timer-Name": t.Name, + } + (*timer.publisher).Publish("", headers, fission.UrlForFunction(&t.Function)) + }) + c.Start() + log.Printf("Updated new cron for %v", t.Name) + return c +} diff --git a/timer/timerSync.go b/timer/timerSync.go new file mode 100644 index 00000000..692d37b9 --- /dev/null +++ b/timer/timerSync.go @@ -0,0 +1,58 @@ +/* +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 timer + +import ( + "log" + "time" + + controllerClient "github.com/fission/fission/controller/client" +) + +type ( + TimerSync struct { + controller *controllerClient.Client + timer *Timer + } +) + +func MakeTimerSync(controller *controllerClient.Client, timer *Timer) *TimerSync { + ws := &TimerSync{ + controller: controller, + timer: timer, + } + go ws.syncSvc() + return ws +} + +func (ws *TimerSync) syncSvc() { + failureCount := 0 + maxFailures := 6 + for { + triggers, err := ws.controller.TimeTriggerList() + if err != nil { + failureCount++ + if failureCount > maxFailures { + log.Fatalf("Failed to connect to controller: %v", err) + } + time.Sleep(10 * time.Second) + continue + } + ws.timer.Sync(triggers) + time.Sleep(3 * time.Second) + } +} diff --git a/types.go b/types.go index 3f7954f5..25b320cf 100644 --- a/types.go +++ b/types.go @@ -67,6 +67,16 @@ type ( Target string `json:"target"` // Watch publish target (URL, NATS stream, etc) } + // TimeTrigger invokes the specific function at a time or + // times specified by a cron string. + TimeTrigger struct { + Metadata `json:"metadata"` + + Cron string `json:"cron"` + + Function Metadata `json:"function"` + } + // Errors returned by the Fission API. Error struct { Code errorCode `json:"code"`