Call poolmgr from router while using a cached service

Poolmgr needs to know usage statistics for a function's pod.  This
change asynchronously taps poolmgr API when router uses a service.
Poolmgr can use this information to control pod expiry.  It may also
be useful later as one of the metrics for autoscaling.

poolmgr/api.go

@@ -23,6 +23,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/gorilla/handlers"
@@ -45,7 +46,8 @@ type funcSvc struct {
 type API struct {
 	poolMgr         *GenericPoolManager
 	functionEnv     *cache.Cache // map[fission.Metadata]fission.Environment
-	functionService *cache.Cache // map[fission.Metadata]funcSvc
+	functionService *cache.Cache // map[fission.Metadata]*funcSvc
+	urlFuncSvc      *cache.Cache // map[string]*funcSvc
 	controller      *controllerclient.Client
 }
 
@@ -54,6 +56,7 @@ func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API
 		poolMgr:         gpm,
 		functionEnv:     cache.MakeCache(),
 		functionService: cache.MakeCache(),
+		urlFuncSvc:      cache.MakeCache(),
 		controller:      controller,
 	}
 }
@@ -120,6 +123,7 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) {
 	if err == nil {
 		//    Ok:     return svc name
 		svc := result.(*funcSvc)
+		svc.atime = time.Now()
 		return svc.serviceName, nil
 	}
 
@@ -151,15 +155,46 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) {
 	err = api.functionService.Set(m, funcSvc)
 	if err != nil {
 		// log and ignore error
-		log.Printf("Error saving function service: %v", err)
+		log.Printf("Error caching function service: %v", err)
+	}
+
+	// cache by svc hostname, for tapService()
+	err = api.urlFuncSvc.Set(funcSvc.serviceName, funcSvc)
+	if err != nil {
+		// log and ignore error
+		log.Printf("Error caching function service by name: %v", err)
 	}
 
 	return funcSvc.serviceName, nil
 }
 
+// find funcSvc and update its atime
+func (api *API) tapService(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "Failed to read request", 500)
+		return
+	}
+	svcName := string(body)
+	svcHost := strings.TrimPrefix(svcName, "http://")
+
+	log.Printf("tap svc: %v", svcHost)
+
+	funcSvcI, err := api.urlFuncSvc.Get(svcHost)
+	if err != nil {
+		http.Error(w, "Not found", 404)
+		return
+	}
+
+	(funcSvcI.(*funcSvc)).atime = time.Now()
+
+	w.WriteHeader(http.StatusOK)
+}
+
 func (api *API) Serve(port int) {
 	r := mux.NewRouter()
 	r.HandleFunc("/v1/getServiceForFunction", api.getServiceForFunctionApi).Methods("POST")
+	r.HandleFunc("/v1/tapService", api.tapService).Methods("POST")
 
 	address := fmt.Sprintf(":%v", port)
 	log.Printf("starting poolmgr at port %v", port)

poolmgr/client/client.go

@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"github.com/platform9/fission"
 	"io/ioutil"
+	"net/url"
 )
 
 type Client struct {
@@ -58,3 +59,19 @@ func (c *Client) GetServiceForFunction(metadata *fission.Metadata) (string, erro
 
 	return string(svcName), nil
 }
+
+func (c *Client) TapService(serviceUrl *url.URL) error {
+	url := c.poolmgrUrl + "/v1/tapService"
+
+	serviceUrlStr := serviceUrl.String()
+
+	resp, err := http.Post(url, "application/octet-stream", bytes.NewReader([]byte(serviceUrlStr)))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		return fission.MakeErrorFromHTTP(resp)
+	}
+	return nil
+}

router/functionHandler.go

@@ -80,7 +80,16 @@ func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
 	return http.DefaultTransport.RoundTrip(req)
 }
 
+func (fh *functionHandler) tapService(serviceUrl *url.URL) {
+	err := fh.poolmgr.TapService(serviceUrl)
+	if err != nil {
+		log.Printf("tap service error: %v", serviceUrl.String())
+	}
+}
+
 func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) {
+
+	// cache lookup
 	serviceUrl, err := fh.fmap.lookup(&fh.Function)
 	if err != nil {
 		// Cache miss: request the Pool Manager to make a new service.
@@ -99,6 +108,10 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
 
 		// add it to the map
 		fh.fmap.assign(&fh.Function, serviceUrl)
+	} else {
+		// if we're using our cache, asynchronously tell
+		// poolmgr we're using this service
+		go fh.tapService(serviceUrl)
 	}
 
 	// Proxy off our request to the serviceUrl, and send the response back.