diff --git a/poolmgr/api.go b/poolmgr/api.go index 9206cc92..3205dc77 100644 --- a/poolmgr/api.go +++ b/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) diff --git a/poolmgr/client/client.go b/poolmgr/client/client.go index 3556f059..f716dda2 100644 --- a/poolmgr/client/client.go +++ b/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 +} diff --git a/router/functionHandler.go b/router/functionHandler.go index ca7c88ff..3d205db9 100644 --- a/router/functionHandler.go +++ b/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.