Add Time Trigger API and client (#153) (#161)

This change adds a timer trigger API, client, and implementation.  Users can trigger a function with a cron-compatible string. 

This change also moves the publisher interface and webhook-based publisher implementation out of kubewatcher and into a separate `publisher` package.

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")

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})
 

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
+}

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(""))
+}

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
+}

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=<port> [--controllerUrl=<url> --namespace=<namespace>]
   fission-bundle --kubewatcher [--controllerUrl=<url> --routerUrl=<url>]
   fission-bundle --logger
+  fission-bundle --timer [--controllerUrl=<url> --routerUrl=<url>]
 Options:
   --controllerPort=<port>  Port that the controller should listen on.
   --routerPort=<port>      Port that the router should listen on.
@@ -99,6 +108,7 @@ Options:
   --namespace=<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 {}
 }

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

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},
 

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
+}

glide.lock

@@ -1,5 +1,5 @@
-hash: 6b4c050c1cd5a7cbb99141348020c882d088b30a9fe9cf92b78a7f706e59da4c
+hash: 51b5eda27974aac228f42002cb0a9f89bf2f79d36a574a3694c142c2f83e4042
-updated: 2017-02-22T23:15:19.862789385+08:00
+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

glide.yaml

@@ -33,3 +33,4 @@ import:
   version: v1.2.0
   subpackages:
   - client/v2
+- package: github.com/robfig/cron

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()

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.MakeClient(controllerUrl), kubeWatch)
-	MakeWatchSync(client, kubeWatch)
 
 	return nil
 }

publisher/publisher.go

@@ -14,17 +14,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kubewatcher
+package publisher
-
-import (
-	"k8s.io/client-go/1.5/pkg/watch"
-)
 
 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)
 	}
 )

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
+		body       string
-		watchEvent watch.Event
+		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,
+		body:       body,
-		url:        url,
+		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.target, "/")
-	url := p.baseUrl + "/" + strings.TrimPrefix(r.url, "/")
 	log.Printf("Making HTTP request to %v", url)
 
-	// Serialize the object
 	var buf bytes.Buffer
-	err := printKubernetesObject(r.watchEvent.Object, &buf)
+	buf.WriteString(r.body)
-	if err != nil {
-		log.Printf("Failed to serialize object: %v", err)
-		// TODO send a POST request indicating error
-	}
 
 	// Create request
 	req, err := http.NewRequest("POST", url, &buf)
-	if err != nil {
+	for k, v := range r.headers {
-		log.Printf("Failed to create request to %v", url)
+		req.Header.Add(k, v)
-		// can't do anything more, drop the event.
-		return
 	}
 
-	// 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)
 

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
 }

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
+}

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
+}

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)
+	}
+}

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"`