V2 types and TPR (#266)

This changes the core fission function, environment and trigger types. It also changes Fission's storage to use ThirdPartyResources.

 - Functions are now specified by packages. Functions can also have both source and deployment packages. A package can be specified by a literal, or by a URL.
 - Environments have a build and runtime component.
 - Triggers reference functions by a FunctionReference. This is a layer of indirection between triggers and functions, and will allow things like incremental function upgrades in future releases.

See Documentation/wip/env-v2.md for design discussion about points 1 and 2.

Changes:

* V2 Types

All types now have a spec, following the pattern of K8s objects.

Functions now have source and deployment packages. A Package can be
specified by literal, or by URL.

Environments now have a builder and runtime component.

All triggers use a new FunctionReference to specify the function. This
for now only uses a function name, but in the future can be extended
to be more flexible.

A new FunctionLoadRequest type is added for specialization requests to
the environment runtime.

* TPR types, TPR init code, and a "fission client"

Implements TPR types using the spec types in fission/types.go.

Adds code for adding creating TPR types, and convenient types for crud
operations on each of our resource types.

Adds code for connecting to K8s API and configuring a REST client with
fission types set up.

* Change old stateful controller into a thin apiserver

This apiserver is now simply a stateless api layer on top of the TPR
types. At the moment it doesn't do anything that couldn't be done by
simply talking to the TPR types. In the future we can have better
validation and potentially some higher level APIs (like versioning for
example) in here.

* Split controller client into files and update for v2 types.

* Update CLI for v2 types.

As far as possible we keep the CLI flags the same. We'll have to add
flags for source/deploy packages and builder/runtime
environments. That will come in the next change.

* Update poolmgr and fetcher for v2 types.

Also adds a poolmgr_test.

* Update router for new types.

Also adds a function reference resolver, which separates out the job
of resolving a FunctionReference to a function.

* Update kubewatcher and timer for v2 types.

* Update Message Queue trigger type for v2 types.

* Minor odds and ends.

* Fission bundle CLI updates

Remove controllerUrl flag, since we don't need it any more.

* Remove etcd deployment (replaced by storing state in TPR)

Also update the poolmgr commandline, and use an env var for the
fetcher image URL.

* Explicit ChecksumType and consts

* Clarify separation of environment interface types

mqtrigger/messageQueue/messageQueue.go

@@ -21,9 +21,10 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"k8s.io/client-go/1.5/pkg/api"
 
-	"github.com/fission/fission"
-	controllerClient "github.com/fission/fission/controller/client"
+	//"github.com/fission/fission"
+	"github.com/fission/fission/tpr"
 )
 
 const (
@@ -47,21 +48,20 @@ type (
 	}
 
 	MessageQueue interface {
-		subscribe(trigger fission.MessageQueueTrigger) (messageQueueSubscription, error)
+		subscribe(trigger *tpr.Messagequeuetrigger) (messageQueueSubscription, error)
 		unsubscribe(triggerSub messageQueueSubscription) error
 	}
 
 	MessageQueueTriggerManager struct {
-		reqChan      chan request
-		mqCfg        MessageQueueConfig
-		triggers     map[string]*triggerSubscription
-		controller   *controllerClient.Client
-		messageQueue MessageQueue
+		reqChan       chan request
+		mqCfg         MessageQueueConfig
+		triggers      map[string]*triggerSubscription
+		fissionClient *tpr.FissionClient
+		messageQueue  MessageQueue
 	}
 
 	triggerSubscription struct {
-		fission.Metadata
-		funcMeta     fission.Metadata
+		trigger      tpr.Messagequeuetrigger
 		subscription messageQueueSubscription
 	}
 
@@ -72,20 +72,18 @@ type (
 	}
 	response struct {
 		err      error
-		triggers *map[string]messageQueueSubscription
+		triggers *map[string]*triggerSubscription
 	}
 )
 
-func MakeMessageQueueTriggerManager(ctrlClient *controllerClient.Client,
-	routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
-
+func MakeMessageQueueTriggerManager(fissionClient *tpr.FissionClient, routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
 	var messageQueue MessageQueue
 	var err error
 
 	mqTriggerMgr := MessageQueueTriggerManager{
-		reqChan:    make(chan request),
-		triggers:   make(map[string]*triggerSubscription),
-		controller: ctrlClient,
+		reqChan:       make(chan request),
+		triggers:      make(map[string]*triggerSubscription),
+		fissionClient: fissionClient,
 	}
 	switch mqConfig.MQType {
 	case NATS:
@@ -108,22 +106,21 @@ func (mqt *MessageQueueTriggerManager) service() {
 		switch req.requestType {
 		case ADD_TRIGGER:
 			var err error
-			triggerUid := req.triggerSub.Uid
-			if _, ok := mqt.triggers[triggerUid]; ok {
+			k := tpr.CacheKey(&req.triggerSub.trigger.Metadata)
+			if _, ok := mqt.triggers[k]; ok {
 				err = errors.New("Trigger already exists")
 			} else {
-				mqt.triggers[triggerUid] = req.triggerSub
+				mqt.triggers[k] = req.triggerSub
 			}
 			req.respChan <- response{err: err}
 		case GET_ALL_TRIGGERS:
-			copyTriggers := make(map[string]messageQueueSubscription)
+			copyTriggers := make(map[string]*triggerSubscription)
 			for key, val := range mqt.triggers {
 				copyTriggers[key] = val
 			}
 			req.respChan <- response{triggers: &copyTriggers}
 		case DELETE_TRIGGER:
-			triggerUid := req.triggerSub.Uid
-			delete(mqt.triggers, triggerUid)
+			delete(mqt.triggers, tpr.CacheKey(&req.triggerSub.trigger.Metadata))
 		}
 	}
 }
@@ -139,7 +136,7 @@ func (mqt *MessageQueueTriggerManager) addTrigger(triggerSub *triggerSubscriptio
 	return r.err
 }
 
-func (mqt *MessageQueueTriggerManager) getAllTriggers() *map[string]messageQueueSubscription {
+func (mqt *MessageQueueTriggerManager) getAllTriggers() *map[string]*triggerSubscription {
 	respChan := make(chan response)
 	mqt.reqChan <- request{
 		requestType: GET_ALL_TRIGGERS,
@@ -149,12 +146,12 @@ func (mqt *MessageQueueTriggerManager) getAllTriggers() *map[string]messageQueue
 	return r.triggers
 }
 
-func (mqt *MessageQueueTriggerManager) delTrigger(triggerUid string) {
+func (mqt *MessageQueueTriggerManager) delTrigger(m *api.ObjectMeta) {
 	mqt.reqChan <- request{
 		requestType: DELETE_TRIGGER,
 		triggerSub: &triggerSubscription{
-			Metadata: fission.Metadata{
-				Uid: triggerUid,
+			trigger: tpr.Messagequeuetrigger{
+				Metadata: *m,
 			},
 		},
 	}
@@ -162,57 +159,61 @@ func (mqt *MessageQueueTriggerManager) delTrigger(triggerUid string) {
 
 func (mqt *MessageQueueTriggerManager) syncTriggers() {
 	for {
-		// TODO: handle error
-		newTriggers, err := mqt.controller.MessageQueueTriggerList(mqt.mqCfg.MQType)
+		// get new set of triggers
+		newTriggers, err := mqt.fissionClient.Messagequeuetriggers(api.NamespaceAll).List(api.ListOptions{})
 		if err != nil {
-			log.Warnf("Failed to sync message queue trigger from controller: %v", err)
+			log.Fatalf("Failed to read message queue trigger list: %v", err)
 		}
-		// sync trigger from controller
-		newTriggerMap := map[string]fission.MessageQueueTrigger{}
-		for _, trigger := range newTriggers {
-			newTriggerMap[trigger.Uid] = trigger
+		newTriggerMap := make(map[string]*tpr.Messagequeuetrigger)
+		for _, trigger := range newTriggers.Items {
+			newTriggerMap[tpr.CacheKey(&trigger.Metadata)] = &trigger
 		}
-		currentTriggerSubMap := mqt.getAllTriggers()
+
+		// get current set of triggers
+		currentTriggers := mqt.getAllTriggers()
 
 		// register new triggers
 		for key, trigger := range newTriggerMap {
-			if _, ok := (*currentTriggerSubMap)[key]; ok {
+			if _, ok := (*currentTriggers)[key]; ok {
 				continue
 			}
+
+			// actually subscribe using the message queue client impl
 			sub, err := mqt.messageQueue.subscribe(trigger)
 			if err != nil {
-				log.Warnf("Message queue trigger %s created failed: %v", trigger.Name, err)
+				log.Warnf("Failed to subscribe to message queue trigger %s: %v", trigger.Metadata.Name, err)
 				continue
 			}
+
 			triggerSub := triggerSubscription{
-				Metadata: fission.Metadata{
-					Name: trigger.Name,
-					Uid:  trigger.Uid,
-				},
-				funcMeta:     trigger.Function,
+				trigger:      *trigger,
 				subscription: sub,
 			}
+
+			// add to our list
 			err = mqt.addTrigger(&triggerSub)
 			if err != nil {
-				log.Warnf("Message queue trigger %s created failed: %v", trigger.Name, err)
-				continue
+				log.Fatalf("Message queue trigger %s addition failed: %v", trigger.Metadata.Name, err)
 			}
-			log.Infof("Message queue trigger %s created", trigger.Name)
+
+			log.Infof("Message queue trigger %s created", trigger.Metadata.Name)
 		}
 
 		// remove old triggers
-		for _, ts := range *currentTriggerSubMap {
-			triggerSub := ts.(*triggerSubscription)
-			if _, ok := newTriggerMap[triggerSub.Uid]; ok {
+		for key, triggerSub := range *currentTriggers {
+			if _, ok := newTriggerMap[key]; ok {
 				continue
 			}
-			if err := mqt.messageQueue.unsubscribe(triggerSub.subscription); err != nil {
-				log.Warnf("Message queue trigger %s deleted failed: %v", triggerSub.Name, err)
-			} else {
-				mqt.delTrigger(triggerSub.Uid)
-				log.Infof("Message queue trigger %s deleted", triggerSub.Name)
+			err := mqt.messageQueue.unsubscribe(triggerSub.subscription)
+			if err != nil {
+				log.Warnf("Failed to unsubscribe to trigger %s: %v", triggerSub.trigger.Metadata.Name, err)
+				continue
 			}
+			mqt.delTrigger(&triggerSub.trigger.Metadata)
+			log.Infof("Message queue trigger %s deleted", triggerSub.trigger.Metadata.Name)
 		}
+
+		// TODO replace with a watch
 		time.Sleep(3 * time.Second)
 	}
 }

mqtrigger/messageQueue/nats.go

@@ -29,6 +29,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/fission/fission"
+	"github.com/fission/fission/tpr"
 )
 
 const (
@@ -57,17 +58,17 @@ func makeNatsMessageQueue(routerUrl string, mqCfg MessageQueueConfig) (MessageQu
 	return nats, nil
 }
 
-func (nats Nats) subscribe(trigger fission.MessageQueueTrigger) (messageQueueSubscription, error) {
-	subj := trigger.Topic
+func (nats Nats) subscribe(trigger *tpr.Messagequeuetrigger) (messageQueueSubscription, error) {
+	subj := trigger.Spec.Topic
 
 	if !isTopicValidForNats(subj) {
-		return nil, errors.New(fmt.Sprintf("Not a valid topic: %s", trigger.Topic))
+		return nil, errors.New(fmt.Sprintf("Not a valid topic: %s", trigger.Spec.Topic))
 	}
 
 	opts := []ns.SubscriptionOption{
 		// Create a durable subscription to nats, so that triggers could retrieve last unack message.
 		// https://github.com/nats-io/go-nats-streaming#durable-subscriptions
-		ns.DurableName(trigger.Uid),
+		ns.DurableName(string(trigger.Metadata.UID)),
 
 		// Nats-streaming server is auto-ack mode by default. Since we want nats-streaming server to
 		// resend a message if the trigger does not ack it, we need to enable the manual ack mode, so that
@@ -90,14 +91,21 @@ func isTopicValidForNats(topic string) bool {
 	return nsUtil.IsSubjectValid(topic)
 }
 
-func msgHandler(nats *Nats, trigger fission.MessageQueueTrigger) func(*ns.Msg) {
+func msgHandler(nats *Nats, trigger *tpr.Messagequeuetrigger) func(*ns.Msg) {
 	return func(msg *ns.Msg) {
-		url := nats.routerUrl + "/" + strings.TrimPrefix(fission.UrlForFunction(&trigger.Function), "/")
+
+		// Support other function ref types
+		if trigger.Spec.FunctionReference.Type != fission.FunctionReferenceTypeFunctionName {
+			log.Fatalf("Unsupported function reference type (%v) for trigger %v",
+				trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
+		}
+
+		url := nats.routerUrl + "/" + strings.TrimPrefix(fission.UrlForFunction(trigger.Spec.FunctionReference.Name), "/")
 		log.Printf("Making HTTP request to %v", url)
 
 		headers := map[string]string{
-			"X-Fission-MQTrigger-Topic":     trigger.Topic,
-			"X-Fission-MQTrigger-RespTopic": trigger.ResponseTopic,
+			"X-Fission-MQTrigger-Topic":     trigger.Spec.Topic,
+			"X-Fission-MQTrigger-RespTopic": trigger.Spec.ResponseTopic,
 		}
 
 		// Create request
@@ -129,10 +137,10 @@ func msgHandler(nats *Nats, trigger fission.MessageQueueTrigger) func(*ns.Msg) {
 			log.Warningf("Failed to ack message: %v", err)
 		}
 
-		if len(trigger.ResponseTopic) > 0 {
-			err = nats.nsConn.Publish(trigger.ResponseTopic, body)
+		if len(trigger.Spec.ResponseTopic) > 0 {
+			err = nats.nsConn.Publish(trigger.Spec.ResponseTopic, body)
 			if err != nil {
-				log.Warningf("Failed to publish message to topic %s: %v", trigger.ResponseTopic, err)
+				log.Warningf("Failed to publish message to topic %s: %v", trigger.Spec.ResponseTopic, err)
 			}
 		}
 	}