Use latest function metadata to check cached function service. (#316)

Watch functions in the router and use that to trigger functionReferenceResolver cache invalidation.

We might want to rate-limit syncTriggers in a future change, since we might be triggering it more often than needed.

router/functionReferenceResolver.go

@@ -116,3 +116,21 @@ func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*re
 	}
 	return &rr, nil
 }
+
+func (frr *functionReferenceResolver) delete(namespace string, fr *fission.FunctionReference) error {
+	nfr := namespacedFunctionReference{
+		namespace:         namespace,
+		functionReference: *fr,
+	}
+	return frr.refCache.Delete(nfr)
+}
+
+func (frr *functionReferenceResolver) copy() map[namespacedFunctionReference]resolveResult {
+	cache := make(map[namespacedFunctionReference]resolveResult)
+	for k, v := range frr.refCache.Copy() {
+		key := k.(namespacedFunctionReference)
+		val := v.(resolveResult)
+		cache[key] = val
+	}
+	return cache
+}

router/httpTriggers.go

@@ -61,6 +61,7 @@ func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
 		return
 	}
 	go ts.watchTriggers()
+	go ts.watchFunctions()
 }
 
 func defaultHomeHandler(w http.ResponseWriter, r *http.Request) {
@@ -162,6 +163,48 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 	}
 }
 
+func (ts *HTTPTriggerSet) watchFunctions() {
+	rv := ""
+	for {
+		wi, err := ts.fissionClient.Functions(api.NamespaceAll).Watch(api.ListOptions{
+			ResourceVersion: rv,
+		})
+		if err != nil {
+			log.Fatalf("Failed to watch function list: %v", err)
+		}
+
+		for {
+			ev, more := <-wi.ResultChan()
+			if !more {
+				// restart watch from last rv
+				break
+			}
+			if ev.Type == watch.Error {
+				// restart watch from the start
+				rv = ""
+				time.Sleep(time.Second)
+				break
+			}
+			fn := ev.Object.(*tpr.Function)
+			rv = fn.Metadata.ResourceVersion
+
+			// update resolver function reference cache
+			for key, rr := range ts.resolver.copy() {
+				if key.functionReference.Name == fn.Metadata.Name &&
+					rr.functionMetadata.ResourceVersion != fn.Metadata.ResourceVersion {
+					err := ts.resolver.delete(key.namespace, &key.functionReference)
+					if err != nil {
+						log.Printf("Error deleting functionReferenceResolver cache: %v", err)
+					}
+					break
+				}
+			}
+
+			ts.syncTriggers()
+		}
+	}
+}
+
 func (ts *HTTPTriggerSet) syncTriggers() {
 	log.Printf("Syncing http triggers")