Soam Vasani
7d1058d7be
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)
94 lines
2.1 KiB
Go
94 lines
2.1 KiB
Go
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)
|
|
}
|