From 7d1058d7be13944b4295ce9b7967fdf984efba17 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Mon, 31 Oct 2016 02:32:16 -0700 Subject: [PATCH 1/3] Pool Manager -- manage generic containers and their specialization GenericPool is a pool of generic containers for an environment. GenericPoolManager keeps track of all GenericPools, creating them on-demand. The pool manager API is simply a "lookup" for the service URL of a function. If one exists it is returned immediately; otherwise, a generic pool is created, and then a pod is specialized from that pool. poolmgr is designed to run from within the cluster, since it connects to pod IP addresses directly. This is a first cut with many pieces missing. TODO: * Use versioned kubernetes clients instead of the unversioned one * Unit tests for GenericPoolMgr; improve unit test for GenericPool; test for API. * Handle cases where a service exists but pod backing it has failed. * On start up, use existing deployments/pods/services if they exist; in other words don't orphan resources on restart. * Kill idle resources (services, pods, even generic pools) * Autoscale generic pool (for example, by watching num ready pods) --- poolmgr/api.go | 138 +++++++++++++++++ poolmgr/gp.go | 372 +++++++++++++++++++++++++++++++++++++++++++++ poolmgr/gp_test.go | 93 ++++++++++++ poolmgr/gpm.go | 77 ++++++++++ 4 files changed, 680 insertions(+) create mode 100644 poolmgr/api.go create mode 100644 poolmgr/gp.go create mode 100644 poolmgr/gp_test.go create mode 100644 poolmgr/gpm.go diff --git a/poolmgr/api.go b/poolmgr/api.go new file mode 100644 index 00000000..e7c0e4f1 --- /dev/null +++ b/poolmgr/api.go @@ -0,0 +1,138 @@ +/* +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 poolmgr + +import ( + "encoding/json" + "net/http" + "time" + + "github.com/platform9/fission" + "github.com/platform9/fission/cache" + controllerclient "github.com/platform9/fission/controller/client" + "io/ioutil" + "log" +) + +type funcSvc struct { + function *fission.Metadata // function this thing is for + environment *fission.Environment // env it was obtained from + serviceName string // name of k8s svc + + ctime time.Time + atime time.Time +} + +type API struct { + poolMgr *GenericPoolManager + functionEnv *cache.Cache // map[fission.Metadata]fission.Environment + functionService *cache.Cache // map[fission.Metadata]funcSvc + controller *controllerclient.Client +} + +func (api *API) lookupApi(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request", 500) + return + } + + // get function metadata + m := fission.Metadata{} + err = json.Unmarshal(body, &m) + if err != nil { + http.Error(w, "Failed to parse request", 400) + return + } + + serviceUrl, err := api.lookup(&m) + if err != nil { + code, msg := fission.GetHTTPError(err) + log.Printf("Error: %v: %v", code, msg) + http.Error(w, msg, code) + } + + // return serviceUrl + w.Write([]byte(serviceUrl)) +} + +func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error) { + var env *fission.Environment + + // Cached ? + result, err := api.functionEnv.Get(m) + if err == nil { + env = result.(*fission.Environment) + return env, nil + } + + // Cache miss -- get func from controller + f, err := api.controller.FunctionGet(m) + if err != nil { + return nil, err + } + + // Get env from metadata + env, err = api.controller.EnvironmentGet(&f.Environment) + if err != nil { + return nil, err + } + + // cache for future + api.functionEnv.Set(m, env) + + return env, nil +} + +func (api *API) lookup(m *fission.Metadata) (string, error) { + // Check function -> svc map + result, err := api.functionService.Get(m) + if err == nil { + // Ok: return svc name + svc := result.(*funcSvc) + return svc.serviceName, nil + } + + // None exists, so create a new funcSvc: + + // from Func -> get Env + env, err := api.getFunctionEnv(m) + if err != nil { + return "", err + } + + // from Env -> get GenericPool + pool, err := api.poolMgr.GetPool(env) + if err != nil { + return "", err + } + + // from GenericPool -> get one function container + funcSvc, err := pool.GetFuncSvc(m) + if err != nil { + return "", err + } + + // add to cache + err = api.functionService.Set(m, funcSvc) + if err != nil { + // log and ignore error + log.Printf("Error saving function service: %v", err) + } + + return funcSvc.serviceName, nil +} diff --git a/poolmgr/gp.go b/poolmgr/gp.go new file mode 100644 index 00000000..59bfa1c4 --- /dev/null +++ b/poolmgr/gp.go @@ -0,0 +1,372 @@ +/* +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 poolmgr + +import ( + "bytes" + "errors" + "fmt" + "log" + "math/rand" + "net/http" + "time" + + "github.com/platform9/fission" + + "k8s.io/kubernetes/pkg/api" + apiUnversioned "k8s.io/kubernetes/pkg/api/unversioned" + "k8s.io/kubernetes/pkg/apis/extensions" + clientUnversioned "k8s.io/kubernetes/pkg/client/unversioned" + "k8s.io/kubernetes/pkg/labels" +) + +type ( + GenericPool struct { + env *fission.Environment + replicas int // num containers + deployment *extensions.Deployment // kubernetes deployment + namespace string // namespace to keep our resources + podReadyTimeout time.Duration // timeout for generic pods to become ready + controllerHostName string + + kubernetesClient *clientUnversioned.Client + requestChannel chan *choosePodRequest + } + + // serialize the choosing of pods so that choices don't conflict + choosePodRequest struct { + newLabels map[string]string + responseChannel chan *choosePodResponse + } + choosePodResponse struct { + pod *api.Pod + error + } +) + +func MakeGenericPool( + kubernetesClient *clientUnversioned.Client, + env *fission.Environment, + initialReplicas int, + namespace string) (*GenericPool, error) { + + gp := &GenericPool{ + env: env, + replicas: initialReplicas, + requestChannel: make(chan *choosePodRequest), + kubernetesClient: kubernetesClient, + namespace: namespace, + podReadyTimeout: 5 * time.Minute, + controllerHostName: "controller", + } + + // create the pool + err := gp.createPool() + if err != nil { + return nil, err + } + + // wait for at least one pod to be ready + err = gp.waitForReadyPod() + if err != nil { + return nil, err + } + + go gp.choosePodService() + return gp, nil +} + +// choosePodService serializes the choosing of pods +func (gp *GenericPool) choosePodService() { + for { + select { + case req := <-gp.requestChannel: + pod, err := gp._choosePod(req.newLabels) + if err != nil { + req.responseChannel <- &choosePodResponse{error: err} + continue + } + req.responseChannel <- &choosePodResponse{pod: pod} + } + } +} + +// choosePod picks a ready pod from the pool and relabels it, waiting if necessary. +// returns the pod API object. +func (gp *GenericPool) choosePod(newLabels map[string]string) (*api.Pod, error) { + req := &choosePodRequest{ + newLabels: newLabels, + responseChannel: make(chan *choosePodResponse), + } + gp.requestChannel <- req + resp := <-req.responseChannel + return resp.pod, resp.error +} + +// _choosePod is called serially by choosePodService +func (gp *GenericPool) _choosePod(newLabels map[string]string) (*api.Pod, error) { + startTime := time.Now() + for { + // Retries took too long, error out. + if time.Now().Sub(startTime) > gp.podReadyTimeout { + return nil, errors.New("timeout: waited too long to get a ready pod") + } + + // Get pods; filter the ones that are ready + podList, err := gp.kubernetesClient.Pods(gp.namespace).List( + api.ListOptions{ + LabelSelector: labels.Set( + gp.deployment.Spec.Selector.MatchLabels).AsSelector(), + }) + if err != nil { + return nil, err + } + readyPods := make([]api.Pod, len(podList.Items)) + for _, pod := range podList.Items { + podReady := true + for _, cs := range pod.Status.ContainerStatuses { + podReady = podReady && cs.Ready + } + if podReady { + readyPods = append(readyPods, pod) + } + } + + // If there are no ready pods, wait and retry. + if len(readyPods) == 0 { + err = gp.waitForReadyPod() + if err != nil { + return nil, err + } + continue + } + + // Pick a ready pod. For now just choose randomly; + // ideally we'd care about which node it's running on, + // and make a good scheduling decision. + chosenPod := readyPods[rand.Intn(len(readyPods))] + + // Relabel. If the pod already got picked and + // modified, this should fail; in that case just + // retry. + chosenPod.ObjectMeta.Labels = newLabels + _, err = gp.kubernetesClient.Pods(gp.namespace).Update(&chosenPod) + if err != nil { + log.Printf("failed to relabel pod: %v", err) + continue + } + log.Printf("Chose a pod: %v", chosenPod.ObjectMeta.Name) + return &chosenPod, nil + } +} + +func labelsForMetadata(metadata *fission.Metadata) map[string]string { + return map[string]string{ + "functionName": metadata.Name, + "functionUid": metadata.Uid, + } +} + +// specializePod chooses a pod, copies the required user-defined function to that pod +// (via fetcher), and calls the function-run container to load it, resulting in a +// specialized pod. +func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*api.Pod, error) { + newLabels := labelsForMetadata(metadata) + + pod, err := gp.choosePod(newLabels) + if err != nil { + return nil, err + } + + // for fetcher we don't need to create a service, just talk to the pod directly + podIP := pod.Status.PodIP + if len(podIP) == 0 { + return nil, errors.New("Pod has no IP") + } + + // tell fetcher to get the function + fetcherUrl := fmt.Sprintf("http://%v:8000/", podIP) + functionUrl := fmt.Sprintf("http://%v/v1/functions/%v?uid=%v&raw=1", + gp.controllerHostName, metadata.Name, metadata.Uid) + fetcherRequest := fmt.Sprintf("{\"url\": \"%v\", \"filename\": \"user\"}", functionUrl) + + resp, err := http.Post(fetcherUrl, "application/json", bytes.NewReader([]byte(fetcherRequest))) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return nil, errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status)) + } + + // get function run container to specialize + specializeUrl := fmt.Sprintf("http://%v:8888/specialize", podIP) + resp2, err := http.Post(specializeUrl, "", bytes.NewReader([]byte{})) + if err != nil { + return nil, err + } + resp2.Body.Close() + return pod, nil +} + +// A pool is a deployment of generic containers for an env. This +// creates the pool but doesn't wait for any pods to be ready. +func (gp *GenericPool) createPool() error { + poolDeploymentName := fmt.Sprintf("deployment-%v-%v-0", + gp.env.Metadata.Name, gp.env.Metadata.Uid) + + podLabels := map[string]string{ + "pool": poolDeploymentName, + } + + sharedMountPath := "/userfunc" + deployment := &extensions.Deployment{ + ObjectMeta: api.ObjectMeta{ + Name: poolDeploymentName, + Labels: map[string]string{ + "environmentName": gp.env.Metadata.Name, + "environmentUid": gp.env.Metadata.Uid, + }, + }, + Spec: extensions.DeploymentSpec{ + Replicas: int32(gp.replicas), + Selector: &apiUnversioned.LabelSelector{ + MatchLabels: podLabels, + }, + Template: api.PodTemplateSpec{ + ObjectMeta: api.ObjectMeta{ + Labels: podLabels, + }, + Spec: api.PodSpec{ + Volumes: []api.Volume{ + api.Volume{ + Name: "userfunc", + VolumeSource: api.VolumeSource{ + EmptyDir: &api.EmptyDirVolumeSource{}, + }, + }, + }, + Containers: []api.Container{ + api.Container{ + Name: gp.env.Metadata.Name, + Image: gp.env.RunContainerImageUrl, + ImagePullPolicy: api.PullIfNotPresent, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []api.VolumeMount{ + api.VolumeMount{ + Name: "userfunc", + MountPath: sharedMountPath, + }, + }, + }, + api.Container{ + Name: "fetcher", + Image: "fission/fetcher", + ImagePullPolicy: api.PullIfNotPresent, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []api.VolumeMount{ + api.VolumeMount{ + Name: "userfunc", + MountPath: sharedMountPath, + }, + }, + Command: []string{"/fetcher", sharedMountPath}, + }, + }, + }, + }, + }, + } + depl, err := gp.kubernetesClient.ExtensionsClient.Deployments(gp.namespace).Create(deployment) + if err != nil { + return err + } + gp.deployment = depl + return nil +} + +func (gp *GenericPool) waitForReadyPod() error { + startTime := time.Now() + for { + // TODO: for now we just poll; use a watch instead + depl, err := gp.kubernetesClient.ExtensionsClient.Deployments(gp.namespace).Get(gp.deployment.ObjectMeta.Name) + if err != nil { + log.Printf("err: %v", err) + return err + } + gp.deployment = depl + if gp.deployment.Status.AvailableReplicas > 0 { + return nil + } + + if time.Now().Sub(startTime) > gp.podReadyTimeout { + return errors.New("timeout: waited too long for pod to be ready") + } + time.Sleep(1000 * time.Millisecond) + } +} + +func (gp *GenericPool) createSvc(name string, labels map[string]string) (*api.Service, error) { + service := api.Service{ + ObjectMeta: api.ObjectMeta{ + Name: name, + }, + Spec: api.ServiceSpec{ + Type: api.ServiceTypeClusterIP, + Ports: []api.ServicePort{ + api.ServicePort{ + Protocol: api.ProtocolTCP, + Port: 8888, + }, + }, + Selector: labels, + }, + } + svc, err := gp.kubernetesClient.Services(gp.namespace).Create(&service) + return svc, err +} + +func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { + pod, err := gp.specializePod(m) + if err != nil { + return nil, err + } + log.Printf("Specialized pod: %v", pod.ObjectMeta.Name) + + svcName := fmt.Sprintf("svc-%v", m.Name) + if len(m.Uid) > 0 { + svcName += ("-" + m.Uid) + } + + labels := labelsForMetadata(m) + svc, err := gp.createSvc(svcName, labels) + if err != nil { + return nil, err + } + if svc.ObjectMeta.Name != svcName { + return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name)) + } + + fsvc := &funcSvc{ + function: m, + environment: gp.env, + serviceName: svcName, + ctime: time.Now(), + atime: time.Now(), + } + return fsvc, nil +} diff --git a/poolmgr/gp_test.go b/poolmgr/gp_test.go new file mode 100644 index 00000000..90b1c478 --- /dev/null +++ b/poolmgr/gp_test.go @@ -0,0 +1,93 @@ +package poolmgr + +import ( + "github.com/platform9/fission" + + "fmt" + "k8s.io/kubernetes/pkg/api" + "k8s.io/kubernetes/pkg/client/unversioned" + "k8s.io/kubernetes/pkg/client/unversioned/clientcmd" + "log" + "net/http" + "testing" +) + +func getKubeClient() *unversioned.Client { + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + configOverrides := &clientcmd.ConfigOverrides{} + kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) + config, err := kubeConfig.ClientConfig() + if err != nil { + panic("failed loading client config") + } + client := unversioned.NewOrDie(config) + return client +} + +type staticHandler struct { + resp string +} + +func (s *staticHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(s.resp)) +} + +// staticHttpServer starts an http server at port and responds to any +// request with the given response. Use this to mock the controller +// raw function fetch HTTP endpoint. +func staticHttpServer(port int, response string) { + s := &staticHandler{resp: response} + log.Fatal(http.ListenAndServe(fmt.Sprintf(":%v", port), s)) +} + +func TestGenericPool(t *testing.T) { + namespace := "fission-test" + + client := getKubeClient() + + _, err := client.Namespaces().Create(&api.Namespace{ + ObjectMeta: api.ObjectMeta{ + Name: namespace, + Labels: map[string]string{}, + }, + }) + if err != nil { + log.Panicf("failed to create namespace: %v", err) + } + + // destroys everything in the namespace + defer client.Namespaces().Delete(namespace) + + env := &fission.Environment{ + Metadata: fission.Metadata{ + Name: "test-env", + Uid: "", + }, + RunContainerImageUrl: "fission/testing", + } + + gp, err := MakeGenericPool(client, env, 3, namespace) + if err != nil { + log.Panicf("failed to make generic pool: %v", err) + } + log.Printf("Pool created") + + // test specialization + + testFunc := ` +module.exports = function (context, callback) { + callback(200, "Hello, world!"); +} +` + go staticHttpServer(2222, testFunc) + + m := fission.Metadata{ + Name: "foo", + Uid: "xxx-yyy", + } + fsvc, err := gp.GetFuncSvc(&m) + if err != nil { + log.Fatalf("Error getting function svc: %v", err) + } + log.Printf("fsvc: %v", fsvc) +} diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go new file mode 100644 index 00000000..49799bcb --- /dev/null +++ b/poolmgr/gpm.go @@ -0,0 +1,77 @@ +/* +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 poolmgr + +import ( + "github.com/platform9/fission" + clientUnversioned "k8s.io/kubernetes/pkg/client/unversioned" +) + +type ( + GenericPoolManager struct { + pools map[fission.Environment]*GenericPool + kubernetesClient *clientUnversioned.Client + namespace string + + requestChannel chan *request + } + request struct { + env *fission.Environment + responseChannel chan *response + } + response struct { + error + pool *GenericPool + } +) + +func MakeGenericPoolManager(client *clientUnversioned.Client, namespace string) *GenericPoolManager { + gpm := &GenericPoolManager{ + pools: make(map[fission.Environment]*GenericPool), + kubernetesClient: client, + namespace: namespace, + requestChannel: make(chan *request), + } + go gpm.service() + return gpm +} + +func (gpm *GenericPoolManager) service() { + for { + select { + case req := <-gpm.requestChannel: + pool, ok := gpm.pools[*req.env] + if !ok { + pool, err := MakeGenericPool(gpm.kubernetesClient, req.env, 3, gpm.namespace) + if err != nil { + req.responseChannel <- &response{error: err} + continue + } + gpm.pools[*req.env] = pool + req.responseChannel <- &response{pool: pool} + } + req.responseChannel <- &response{pool: pool} + } + } +} + +func (gpm *GenericPoolManager) GetPool(env *fission.Environment) (*GenericPool, error) { + c := make(chan *response) + gpm.requestChannel <- &request{env: env, responseChannel: c} + resp := <-c + return resp.pool, resp.error +} From 40dfba1a410cb2bd974a1a212f94dab659aedf6d Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 01:17:59 -0700 Subject: [PATCH 2/3] Switch to official Kubernetes Go client -- client-go/1.4 Switch to client-go package instead of pulling the from kubernetes. Use a versioned client with sensible compatiblity. This change breaks 'go get'. For now you have to manually checkout the 'release-1.4' branch of the client-go package after 'go get' fetches it. TODO: use one of the build tools to fix this. --- poolmgr/api.go | 27 ++++++++++- poolmgr/gp.go | 127 +++++++++++++++++++++++++------------------------ poolmgr/gpm.go | 12 +++-- 3 files changed, 96 insertions(+), 70 deletions(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index e7c0e4f1..d457650d 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -18,14 +18,19 @@ package poolmgr import ( "encoding/json" + "fmt" + "io/ioutil" + "log" "net/http" + "os" "time" + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + "github.com/platform9/fission" "github.com/platform9/fission/cache" controllerclient "github.com/platform9/fission/controller/client" - "io/ioutil" - "log" ) type funcSvc struct { @@ -44,6 +49,15 @@ type API struct { controller *controllerclient.Client } +func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API { + return &API{ + poolMgr: gpm, + functionEnv: cache.MakeCache(), + functionService: cache.MakeCache(), + controller: controller, + } +} + func (api *API) lookupApi(w http.ResponseWriter, r *http.Request) { body, err := ioutil.ReadAll(r.Body) if err != nil { @@ -136,3 +150,12 @@ func (api *API) lookup(m *fission.Metadata) (string, error) { return funcSvc.serviceName, nil } + +func (api *API) Serve(port int) { + r := mux.NewRouter() + r.HandleFunc("/v1/lookup", api.lookupApi).Methods("GET") + + address := fmt.Sprintf(":%v", port) + log.Printf("starting poolmgr at port %v", port) + log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) +} diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 59bfa1c4..cb1919ee 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -25,25 +25,25 @@ import ( "net/http" "time" - "github.com/platform9/fission" + "k8s.io/client-go/1.4/kubernetes" + "k8s.io/client-go/1.4/pkg/api" + "k8s.io/client-go/1.4/pkg/api/v1" + "k8s.io/client-go/1.4/pkg/apis/extensions/v1beta1" + "k8s.io/client-go/1.4/pkg/labels" - "k8s.io/kubernetes/pkg/api" - apiUnversioned "k8s.io/kubernetes/pkg/api/unversioned" - "k8s.io/kubernetes/pkg/apis/extensions" - clientUnversioned "k8s.io/kubernetes/pkg/client/unversioned" - "k8s.io/kubernetes/pkg/labels" + "github.com/platform9/fission" ) type ( GenericPool struct { - env *fission.Environment - replicas int // num containers - deployment *extensions.Deployment // kubernetes deployment - namespace string // namespace to keep our resources - podReadyTimeout time.Duration // timeout for generic pods to become ready - controllerHostName string + env *fission.Environment + replicas int32 // num containers + deployment *v1beta1.Deployment // kubernetes deployment + namespace string // namespace to keep our resources + podReadyTimeout time.Duration // timeout for generic pods to become ready + controllerUrl string - kubernetesClient *clientUnversioned.Client + kubernetesClient *kubernetes.Clientset requestChannel chan *choosePodRequest } @@ -53,25 +53,26 @@ type ( responseChannel chan *choosePodResponse } choosePodResponse struct { - pod *api.Pod + pod *v1.Pod error } ) func MakeGenericPool( - kubernetesClient *clientUnversioned.Client, + controllerUrl string, + kubernetesClient *kubernetes.Clientset, env *fission.Environment, - initialReplicas int, + initialReplicas int32, namespace string) (*GenericPool, error) { gp := &GenericPool{ - env: env, - replicas: initialReplicas, - requestChannel: make(chan *choosePodRequest), - kubernetesClient: kubernetesClient, - namespace: namespace, - podReadyTimeout: 5 * time.Minute, - controllerHostName: "controller", + env: env, + replicas: initialReplicas, + requestChannel: make(chan *choosePodRequest), + kubernetesClient: kubernetesClient, + namespace: namespace, + podReadyTimeout: 5 * time.Minute, + controllerUrl: controllerUrl, } // create the pool @@ -107,7 +108,7 @@ func (gp *GenericPool) choosePodService() { // choosePod picks a ready pod from the pool and relabels it, waiting if necessary. // returns the pod API object. -func (gp *GenericPool) choosePod(newLabels map[string]string) (*api.Pod, error) { +func (gp *GenericPool) choosePod(newLabels map[string]string) (*v1.Pod, error) { req := &choosePodRequest{ newLabels: newLabels, responseChannel: make(chan *choosePodResponse), @@ -118,7 +119,7 @@ func (gp *GenericPool) choosePod(newLabels map[string]string) (*api.Pod, error) } // _choosePod is called serially by choosePodService -func (gp *GenericPool) _choosePod(newLabels map[string]string) (*api.Pod, error) { +func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error) { startTime := time.Now() for { // Retries took too long, error out. @@ -127,7 +128,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*api.Pod, error) } // Get pods; filter the ones that are ready - podList, err := gp.kubernetesClient.Pods(gp.namespace).List( + podList, err := gp.kubernetesClient.Core().Pods(gp.namespace).List( api.ListOptions{ LabelSelector: labels.Set( gp.deployment.Spec.Selector.MatchLabels).AsSelector(), @@ -135,7 +136,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*api.Pod, error) if err != nil { return nil, err } - readyPods := make([]api.Pod, len(podList.Items)) + readyPods := make([]v1.Pod, len(podList.Items)) for _, pod := range podList.Items { podReady := true for _, cs := range pod.Status.ContainerStatuses { @@ -164,7 +165,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*api.Pod, error) // modified, this should fail; in that case just // retry. chosenPod.ObjectMeta.Labels = newLabels - _, err = gp.kubernetesClient.Pods(gp.namespace).Update(&chosenPod) + _, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(&chosenPod) if err != nil { log.Printf("failed to relabel pod: %v", err) continue @@ -184,7 +185,7 @@ func labelsForMetadata(metadata *fission.Metadata) map[string]string { // specializePod chooses a pod, copies the required user-defined function to that pod // (via fetcher), and calls the function-run container to load it, resulting in a // specialized pod. -func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*api.Pod, error) { +func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error) { newLabels := labelsForMetadata(metadata) pod, err := gp.choosePod(newLabels) @@ -200,8 +201,8 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*api.Pod, erro // tell fetcher to get the function fetcherUrl := fmt.Sprintf("http://%v:8000/", podIP) - functionUrl := fmt.Sprintf("http://%v/v1/functions/%v?uid=%v&raw=1", - gp.controllerHostName, metadata.Name, metadata.Uid) + functionUrl := fmt.Sprintf("%v/v1/functions/%v?uid=%v&raw=1", + gp.controllerUrl, metadata.Name, metadata.Uid) fetcherRequest := fmt.Sprintf("{\"url\": \"%v\", \"filename\": \"user\"}", functionUrl) resp, err := http.Post(fetcherUrl, "application/json", bytes.NewReader([]byte(fetcherRequest))) @@ -234,52 +235,52 @@ func (gp *GenericPool) createPool() error { } sharedMountPath := "/userfunc" - deployment := &extensions.Deployment{ - ObjectMeta: api.ObjectMeta{ + deployment := &v1beta1.Deployment{ + ObjectMeta: v1.ObjectMeta{ Name: poolDeploymentName, Labels: map[string]string{ "environmentName": gp.env.Metadata.Name, "environmentUid": gp.env.Metadata.Uid, }, }, - Spec: extensions.DeploymentSpec{ - Replicas: int32(gp.replicas), - Selector: &apiUnversioned.LabelSelector{ + Spec: v1beta1.DeploymentSpec{ + Replicas: &gp.replicas, + Selector: &v1beta1.LabelSelector{ MatchLabels: podLabels, }, - Template: api.PodTemplateSpec{ - ObjectMeta: api.ObjectMeta{ + Template: v1.PodTemplateSpec{ + ObjectMeta: v1.ObjectMeta{ Labels: podLabels, }, - Spec: api.PodSpec{ - Volumes: []api.Volume{ - api.Volume{ + Spec: v1.PodSpec{ + Volumes: []v1.Volume{ + v1.Volume{ Name: "userfunc", - VolumeSource: api.VolumeSource{ - EmptyDir: &api.EmptyDirVolumeSource{}, + VolumeSource: v1.VolumeSource{ + EmptyDir: &v1.EmptyDirVolumeSource{}, }, }, }, - Containers: []api.Container{ - api.Container{ + Containers: []v1.Container{ + v1.Container{ Name: gp.env.Metadata.Name, Image: gp.env.RunContainerImageUrl, - ImagePullPolicy: api.PullIfNotPresent, + ImagePullPolicy: v1.PullIfNotPresent, TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []api.VolumeMount{ - api.VolumeMount{ + VolumeMounts: []v1.VolumeMount{ + v1.VolumeMount{ Name: "userfunc", MountPath: sharedMountPath, }, }, }, - api.Container{ + v1.Container{ Name: "fetcher", Image: "fission/fetcher", - ImagePullPolicy: api.PullIfNotPresent, + ImagePullPolicy: v1.PullIfNotPresent, TerminationMessagePath: "/dev/termination-log", - VolumeMounts: []api.VolumeMount{ - api.VolumeMount{ + VolumeMounts: []v1.VolumeMount{ + v1.VolumeMount{ Name: "userfunc", MountPath: sharedMountPath, }, @@ -291,7 +292,7 @@ func (gp *GenericPool) createPool() error { }, }, } - depl, err := gp.kubernetesClient.ExtensionsClient.Deployments(gp.namespace).Create(deployment) + depl, err := gp.kubernetesClient.Extensions().Deployments(gp.namespace).Create(deployment) if err != nil { return err } @@ -303,7 +304,7 @@ func (gp *GenericPool) waitForReadyPod() error { startTime := time.Now() for { // TODO: for now we just poll; use a watch instead - depl, err := gp.kubernetesClient.ExtensionsClient.Deployments(gp.namespace).Get(gp.deployment.ObjectMeta.Name) + depl, err := gp.kubernetesClient.Extensions().Deployments(gp.namespace).Get(gp.deployment.ObjectMeta.Name) if err != nil { log.Printf("err: %v", err) return err @@ -320,23 +321,23 @@ func (gp *GenericPool) waitForReadyPod() error { } } -func (gp *GenericPool) createSvc(name string, labels map[string]string) (*api.Service, error) { - service := api.Service{ - ObjectMeta: api.ObjectMeta{ +func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Service, error) { + service := v1.Service{ + ObjectMeta: v1.ObjectMeta{ Name: name, }, - Spec: api.ServiceSpec{ - Type: api.ServiceTypeClusterIP, - Ports: []api.ServicePort{ - api.ServicePort{ - Protocol: api.ProtocolTCP, + Spec: v1.ServiceSpec{ + Type: v1.ServiceTypeClusterIP, + Ports: []v1.ServicePort{ + v1.ServicePort{ + Protocol: v1.ProtocolTCP, Port: 8888, }, }, Selector: labels, }, } - svc, err := gp.kubernetesClient.Services(gp.namespace).Create(&service) + svc, err := gp.kubernetesClient.Core().Services(gp.namespace).Create(&service) return svc, err } diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 49799bcb..876a39f2 100644 --- a/poolmgr/gpm.go +++ b/poolmgr/gpm.go @@ -18,14 +18,15 @@ package poolmgr import ( "github.com/platform9/fission" - clientUnversioned "k8s.io/kubernetes/pkg/client/unversioned" + "k8s.io/client-go/1.4/kubernetes" ) type ( GenericPoolManager struct { pools map[fission.Environment]*GenericPool - kubernetesClient *clientUnversioned.Client + kubernetesClient *kubernetes.Clientset namespace string + controllerUrl string requestChannel chan *request } @@ -39,11 +40,12 @@ type ( } ) -func MakeGenericPoolManager(client *clientUnversioned.Client, namespace string) *GenericPoolManager { +func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.Clientset, namespace string) *GenericPoolManager { gpm := &GenericPoolManager{ pools: make(map[fission.Environment]*GenericPool), - kubernetesClient: client, + kubernetesClient: kubernetesClient, namespace: namespace, + controllerUrl: controllerUrl, requestChannel: make(chan *request), } go gpm.service() @@ -56,7 +58,7 @@ func (gpm *GenericPoolManager) service() { case req := <-gpm.requestChannel: pool, ok := gpm.pools[*req.env] if !ok { - pool, err := MakeGenericPool(gpm.kubernetesClient, req.env, 3, gpm.namespace) + pool, err := MakeGenericPool(gpm.controllerUrl, gpm.kubernetesClient, req.env, 3, gpm.namespace) if err != nil { req.responseChannel <- &response{error: err} continue From 01bfa8b082b2fd3f885425d8bcbcdbc50e6a51fe Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 01:21:36 -0700 Subject: [PATCH 3/3] Minor changes; remove noisy test logs --- controller/api_test.go | 10 +++++----- controller/client/client.go | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/controller/api_test.go b/controller/api_test.go index 988971b5..961343cf 100644 --- a/controller/api_test.go +++ b/controller/api_test.go @@ -54,7 +54,7 @@ func TestFunctionApi(t *testing.T) { m, err := g.client.FunctionCreate(testFunc) panicIf(err) uid1 := m.Uid - log.Printf("Created function %v: %v", m.Name, m.Uid) + //log.Printf("Created function %v: %v", m.Name, m.Uid) code, err := g.client.FunctionGetRaw(m) panicIf(err) @@ -64,7 +64,7 @@ func TestFunctionApi(t *testing.T) { m, err = g.client.FunctionUpdate(testFunc) panicIf(err) uid2 := m.Uid - log.Printf("Updated function %v: %v", m.Name, m.Uid) + //log.Printf("Updated function %v: %v", m.Name, m.Uid) m.Uid = uid1 testFunc.Code = "code1" @@ -72,8 +72,8 @@ func TestFunctionApi(t *testing.T) { panicIf(err) testFunc.Metadata.Uid = m.Uid - log.Printf("f = %#v", f) - log.Printf("testFunc = %#v", testFunc) + //log.Printf("f = %#v", f) + //log.Printf("testFunc = %#v", testFunc) assert(*f == *testFunc, "first version should match when read by uid") m.Uid = uid2 @@ -212,7 +212,7 @@ func TestMain(m *testing.M) { HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs}, EnvironmentStore: EnvironmentStore{ResourceStore: *rs}, } - g.client = client.New("http://localhost:8888") + g.client = client.MakeClient("http://localhost:8888") ks.Delete(context.Background(), "Function", &etcdClient.DeleteOptions{Recursive: true}) ks.Delete(context.Background(), "HTTPTrigger", &etcdClient.DeleteOptions{Recursive: true}) diff --git a/controller/client/client.go b/controller/client/client.go index cece945b..61a84d39 100644 --- a/controller/client/client.go +++ b/controller/client/client.go @@ -36,7 +36,7 @@ type ( } ) -func New(serverUrl string) *Client { +func MakeClient(serverUrl string) *Client { return &Client{Url: serverUrl} }