diff --git a/buildermgr/common.go b/buildermgr/common.go index c981f233..833242d5 100644 --- a/buildermgr/common.go +++ b/buildermgr/common.go @@ -27,7 +27,6 @@ import ( "github.com/fission/fission/builder" builderClient "github.com/fission/fission/builder/client" "github.com/fission/fission/crd" - "github.com/fission/fission/environments/fetcher" fetcherClient "github.com/fission/fission/environments/fetcher/client" ) @@ -39,7 +38,7 @@ import ( // 4. Return upload response and build logs. // *. Return build logs and error if any one of steps above failed. func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string, - storageSvcUrl string, pkg *crd.Package) (uploadResp *fetcher.UploadResponse, buildLogs string, err error) { + storageSvcUrl string, pkg *crd.Package) (uploadResp *fission.ArchiveUploadResponse, buildLogs string, err error) { env, err := fissionClient.Environments(pkg.Spec.Environment.Namespace).Get(pkg.Spec.Environment.Name) if err != nil { @@ -53,8 +52,8 @@ func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string, fetcherC := fetcherClient.MakeClient(fmt.Sprintf("http://%v:8000", svcName)) builderC := builderClient.MakeClient(fmt.Sprintf("http://%v:8001", svcName)) - fetchReq := &fetcher.FetchRequest{ - FetchType: fetcher.FETCH_SOURCE, + fetchReq := &fission.FunctionFetchRequest{ + FetchType: fission.FETCH_SOURCE, Package: pkg.Metadata, Filename: srcPkgFilename, KeepArchive: false, @@ -96,7 +95,7 @@ func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string, archivePackage := !env.Spec.KeepArchive - uploadReq := &fetcher.UploadRequest{ + uploadReq := &fission.ArchiveUploadRequest{ Filename: buildResp.ArtifactFilename, StorageSvcUrl: storageSvcUrl, ArchivePackage: archivePackage, @@ -117,7 +116,7 @@ func buildPackage(fissionClient *crd.FissionClient, envBuilderNamespace string, func updatePackage(fissionClient *crd.FissionClient, pkg *crd.Package, status fission.BuildStatus, buildLogs string, - uploadResp *fetcher.UploadResponse) (*crd.Package, error) { + uploadResp *fission.ArchiveUploadResponse) (*crd.Package, error) { pkg.Status = fission.PackageStatus{ BuildStatus: status, diff --git a/environments/fetcher/client/client.go b/environments/fetcher/client/client.go index 76bd1964..7b68a707 100644 --- a/environments/fetcher/client/client.go +++ b/environments/fetcher/client/client.go @@ -6,10 +6,10 @@ import ( "io/ioutil" "log" "net/http" + "strings" "time" "github.com/fission/fission" - "github.com/fission/fission/environments/fetcher" ) type ( @@ -20,68 +20,74 @@ type ( func MakeClient(fetcherUrl string) *Client { return &Client{ - url: fetcherUrl, + url: strings.TrimSuffix(fetcherUrl, "/"), } } -func (c *Client) Fetch(fr *fetcher.FetchRequest) error { - body, err := json.Marshal(fr) - if err != nil { - return err - } - - maxRetries := 20 - var resp *http.Response - - for i := 0; i < maxRetries; i++ { - resp, err = http.Post(c.url, "application/json", bytes.NewReader(body)) - - if err == nil { - if resp.StatusCode == 200 { - resp.Body.Close() - return nil - } - err = fission.MakeErrorFromHTTP(resp) - } - - if i < maxRetries-1 { - time.Sleep(50 * time.Duration(2*i) * time.Millisecond) - log.Printf("Error fetching package (%v), retrying", err) - continue - } - - log.Printf("Failed to fetch: %v", err) - return err - } - - return nil +func (c *Client) getSpecializeUrl() string { + return c.url + "/specialize" } -func (c *Client) Upload(fr *fetcher.UploadRequest) (*fetcher.UploadResponse, error) { - body, err := json.Marshal(fr) - if err != nil { - return nil, err - } - resp, err := http.Post(c.url+"/upload", "application/json", bytes.NewReader(body)) - if err != nil { - return nil, err - } - defer resp.Body.Close() +func (c *Client) getFetchUrl() string { + return c.url + "/fetch" +} - if resp.StatusCode != 200 { - return nil, fission.MakeErrorFromHTTP(resp) - } +func (c *Client) getUploadUrl() string { + return c.url + "/upload" +} - rBody, err := ioutil.ReadAll(resp.Body) - if err != nil { - return nil, err - } +func (c *Client) Specialize(req *fission.FunctionSpecializeRequest) error { + _, err := sendRequest(req, c.getSpecializeUrl()) + return err +} - uploadResp := fetcher.UploadResponse{} - err = json.Unmarshal([]byte(rBody), &uploadResp) +func (c *Client) Fetch(fr *fission.FunctionFetchRequest) error { + _, err := sendRequest(fr, c.getFetchUrl()) + return err +} + +func (c *Client) Upload(fr *fission.ArchiveUploadRequest) (*fission.ArchiveUploadResponse, error) { + body, err := sendRequest(fr, c.getUploadUrl()) + + uploadResp := fission.ArchiveUploadResponse{} + err = json.Unmarshal(body, &uploadResp) if err != nil { return nil, err } return &uploadResp, nil } + +func sendRequest(req interface{}, url string) ([]byte, error) { + body, err := json.Marshal(req) + if err != nil { + return nil, err + } + + maxRetries := 20 + var resp *http.Response + + for i := 0; i < maxRetries; i++ { + resp, err = http.Post(url, "application/json", bytes.NewReader(body)) + + if err == nil { + if resp.StatusCode == 200 { + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + log.Printf("Error reading response body: %v", err) + } + resp.Body.Close() + return body, err + } + err = fission.MakeErrorFromHTTP(resp) + } + + if i < maxRetries-1 { + time.Sleep(50 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error specializing/fetching/uploading package (%v) with url %v, retrying", err, url) + continue + } + } + + return nil, err +} diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 8c139074..2e6498f7 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -1,20 +1,15 @@ package main import ( - "bytes" "encoding/json" "flag" "fmt" "log" - "net" "net/http" - "net/url" "os" "os/signal" "runtime/debug" - "strconv" "syscall" - "time" "github.com/fission/fission" "github.com/fission/fission/environments/fetcher" @@ -38,8 +33,7 @@ func main() { flag.Usage = fetcherUsage specializeOnStart := flag.Bool("specialize-on-startup", false, "Flag to activate specialize process at pod starup") - fetchPayload := flag.String("fetch-request", "", "JSON Payload for fetch request") - loadPayload := flag.String("load-request", "", "JSON payload for Load request") + specializePayload := flag.String("specialize-request", "", "JSON payload for specialize request") secretDir := flag.String("secret-dir", "", "Path to shared secrets directory") configDir := flag.String("cfgmap-dir", "", "Path to shared configmap directory") @@ -59,19 +53,44 @@ func main() { } } - fetcher, err := fetcher.MakeFetcher(dir, *secretDir, *configDir) + f, err := fetcher.MakeFetcher(dir, *secretDir, *configDir) if err != nil { log.Fatalf("Error making fetcher: %v", err) } - if *specializeOnStart { - specializePod(fetcher, fetchPayload, loadPayload) - } + readyToServe := false + + // do specialization in other goroutine to prevent blocking in newdeploy + go func() { + if *specializeOnStart { + var specializeReq fission.FunctionSpecializeRequest + + err := json.Unmarshal([]byte(*specializePayload), &specializeReq) + if err != nil { + log.Fatalf("Error decoding specialize request: %v", err) + } + + err = f.SpecializePod(specializeReq.FetchReq, specializeReq.LoadReq) + if err != nil { + log.Fatalf("Error specialing function poadt: %v", err) + } + + readyToServe = true + } + }() mux := http.NewServeMux() - mux.HandleFunc("/", fetcher.FetchHandler) - mux.HandleFunc("/upload", fetcher.UploadHandler) - mux.HandleFunc("/version", fetcher.VersionHandler) + mux.HandleFunc("/fetch", f.FetchHandler) + mux.HandleFunc("/specialize", f.SpecializeHandler) + mux.HandleFunc("/upload", f.UploadHandler) + mux.HandleFunc("/version", f.VersionHandler) + mux.HandleFunc("/readniess-healthz", func(w http.ResponseWriter, r *http.Request) { + if !*specializeOnStart || readyToServe { + w.WriteHeader(http.StatusOK) + } else { + w.WriteHeader(http.StatusServiceUnavailable) + } + }) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) @@ -81,75 +100,5 @@ func main() { } func fetcherUsage() { - fmt.Printf("Usage: fetcher [-specialize-on-startup] [-fetch-request ] [-load-request ] [-secret-dir ] [-cfgmap-dir ] \n") -} - -func specializePod(f *fetcher.Fetcher, fetchPayload *string, loadPayload *string) { - // Fetch code - var fetchReq fetcher.FetchRequest - err := json.Unmarshal([]byte(*fetchPayload), &fetchReq) - if err != nil { - log.Fatalf("Error parsing fetch request: %v", err) - } - _, err = f.Fetch(fetchReq) - if err != nil { - log.Fatalf("Error fetching: %v", err) - } - - _, err = f.FetchSecretsAndCfgMaps(fetchReq.Secrets, fetchReq.ConfigMaps) - if err != nil { - log.Fatalf("Error fetching secrets/configmaps: %v", err) - return - } - - // Specialize the pod - - envVersion, err := strconv.Atoi(os.Getenv("ENV_VERSION")) - if err != nil { - log.Fatalf("Error parsing environment version %v, error: %v", os.Getenv("ENV_VERSION"), err) - } - - maxRetries := 30 - var contentType string - var specializeURL string - var reader *bytes.Reader - - if envVersion >= 2 { - contentType = "application/json" - specializeURL = "http://localhost:8888/v2/specialize" - reader = bytes.NewReader([]byte(*loadPayload)) - } else { - contentType = "text/plain" - specializeURL = "http://localhost:8888/specialize" - reader = bytes.NewReader([]byte{}) - } - - for i := 0; i < maxRetries; i++ { - resp, err := http.Post(specializeURL, contentType, reader) - if err == nil && resp.StatusCode < 300 { - // Success - resp.Body.Close() - break - } - - // Only retry for the specific case of a connection error. - if urlErr, ok := err.(*url.Error); ok { - if netErr, ok := urlErr.Err.(*net.OpError); ok { - if netErr.Op == "dial" { - if i < maxRetries-1 { - time.Sleep(500 * time.Duration(2*i) * time.Millisecond) - log.Printf("Error connecting to pod (%v), retrying", netErr) - continue - } - } - } - } - - if err == nil { - err = fission.MakeErrorFromHTTP(resp) - } - log.Printf("Failed to specialize pod: %v", err) - return - } - + fmt.Printf("Usage: fetcher [-specialize-on-startup] [-specialize-request ] [-secret-dir ] [-cfgmap-dir ] \n") } diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index aa2bf64e..8d3d13d3 100644 --- a/environments/fetcher/fetcher.go +++ b/environments/fetcher/fetcher.go @@ -1,20 +1,23 @@ package fetcher import ( + "bytes" "crypto/sha256" "encoding/hex" "encoding/json" - "errors" "fmt" "io" "io/ioutil" "log" + "net" "net/http" + "net/url" "os" "path/filepath" "time" "github.com/mholt/archiver" + "github.com/pkg/errors" "github.com/satori/go.uuid" k8serr "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -26,34 +29,6 @@ import ( ) type ( - FetchRequestType int - - FetchRequest struct { - FetchType FetchRequestType `json:"fetchType"` - Package metav1.ObjectMeta `json:"package"` - Url string `json:"url"` - StorageSvcUrl string `json:"storagesvcurl"` - Filename string `json:"filename"` - Secrets []fission.SecretReference `json:"secretList"` - ConfigMaps []fission.ConfigMapReference `json:"configMapList"` - KeepArchive bool `json:"keeparchive"` - } - - // UploadRequest send from builder manager describes which - // deployment package should be upload to storage service. - UploadRequest struct { - Filename string `json:"filename"` - StorageSvcUrl string `json:"storagesvcurl"` - ArchivePackage bool `json:"archivepackage"` - } - - // UploadResponse defines the download url of an archive and - // its checksum. - UploadResponse struct { - ArchiveDownloadUrl string `json:"archiveDownloadUrl"` - Checksum fission.Checksum `json:"checksum"` - } - Fetcher struct { sharedVolumePath string sharedSecretPath string @@ -63,12 +38,6 @@ type ( } ) -const ( - FETCH_SOURCE = iota - FETCH_DEPLOYMENT - FETCH_URL // remove this? -) - func makeVolumeDir(dirPath string) { err := os.MkdirAll(dirPath, os.ModeDir|0700) if err != nil { @@ -198,7 +167,7 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) return } - var req FetchRequest + var req fission.FunctionFetchRequest err = json.Unmarshal(body, &req) if err != nil { log.Printf("Error reading request body: %v", err) @@ -225,9 +194,42 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + http.Error(w, fmt.Sprintf("only POST is supported on this endpoint, %v received", r.Method), http.StatusMethodNotAllowed) + return + } + + // parse request + body, err := ioutil.ReadAll(r.Body) + if err != nil { + log.Printf("Error reading request body") + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + var req fission.FunctionSpecializeRequest + err = json.Unmarshal(body, &req) + if err != nil { + log.Printf("Error reading request body: %v", err) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + //log.Printf("fetcher received fetch request and started downloading: %v", req) + + err = fetcher.SpecializePod(req.FetchReq, req.LoadReq) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + + // all done + w.WriteHeader(http.StatusOK) +} + // Fetch takes FetchRequest and makes the fetch call // It returns the HTTP code and error if any -func (fetcher *Fetcher) Fetch(req FetchRequest) (int, error) { +func (fetcher *Fetcher) Fetch(req fission.FunctionFetchRequest) (int, error) { // check that the requested filename is not an empty string and error out if so if len(req.Filename) == 0 { e := fmt.Sprintf("Fetch request received for an empty file name, request: %v", req) @@ -244,7 +246,7 @@ func (fetcher *Fetcher) Fetch(req FetchRequest) (int, error) { tmpFile := req.Filename + ".tmp" tmpPath := filepath.Join(fetcher.sharedVolumePath, tmpFile) - if req.FetchType == FETCH_URL { + if req.FetchType == fission.FETCH_URL { // fetch the file and save it to the tmp path err := downloadUrl(req.Url, tmpPath) if err != nil { @@ -262,9 +264,9 @@ func (fetcher *Fetcher) Fetch(req FetchRequest) (int, error) { } var archive *fission.Archive - if req.FetchType == FETCH_SOURCE { + if req.FetchType == fission.FETCH_SOURCE { archive = &pkg.Spec.Source - } else if req.FetchType == FETCH_DEPLOYMENT { + } else if req.FetchType == fission.FETCH_DEPLOYMENT { // sometimes, the user may invoke the function even before the source code is built into a deploy pkg. // this results in executor sending a fetch request of type FETCH_DEPLOYMENT and since pkg.Spec.Deployment.Url will be empty, // we hit this "Get : unsupported protocol scheme "" error. @@ -418,7 +420,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { return } - var req UploadRequest + var req fission.ArchiveUploadRequest err = json.Unmarshal(body, &req) if err != nil { log.Printf("Error reading request body: %v", err) @@ -468,7 +470,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { return } - resp := UploadResponse{ + resp := fission.ArchiveUploadResponse{ ArchiveDownloadUrl: ssClient.GetUrl(fileID), Checksum: *sum, } @@ -523,3 +525,73 @@ func (fetcher *Fetcher) unarchive(src string, dst string) error { } return nil } + +func (fetcher *Fetcher) SpecializePod(fetchReq fission.FunctionFetchRequest, loadReq fission.FunctionLoadRequest) error { + startTime := time.Now() + defer func() { + elapsed := time.Since(startTime) + log.Printf("Elapsed time in fetch request = %v", elapsed) + }() + + _, err := fetcher.Fetch(fetchReq) + if err != nil { + return errors.Wrap(err, "Error fetching deploy package") + } + + _, err = fetcher.FetchSecretsAndCfgMaps(fetchReq.Secrets, fetchReq.ConfigMaps) + if err != nil { + return errors.Wrap(err, "Error fetching secrets/configmaps") + } + + // Specialize the pod + + maxRetries := 30 + var contentType string + var specializeURL string + var reader *bytes.Reader + + loadPayload, err := json.Marshal(loadReq) + if err != nil { + return errors.Wrap(err, "Error encoding load request") + } + + if loadReq.EnvVersion >= 2 { + contentType = "application/json" + specializeURL = "http://localhost:8888/v2/specialize" + reader = bytes.NewReader(loadPayload) + } else { + contentType = "text/plain" + specializeURL = "http://localhost:8888/specialize" + reader = bytes.NewReader([]byte{}) + } + + for i := 0; i < maxRetries; i++ { + resp, err := http.Post(specializeURL, contentType, reader) + if err == nil && resp.StatusCode < 300 { + // Success + resp.Body.Close() + return nil + } + + // Only retry for the specific case of a connection error. + if urlErr, ok := err.(*url.Error); ok { + if netErr, ok := urlErr.Err.(*net.OpError); ok { + if netErr.Op == "dial" { + if i < maxRetries-1 { + time.Sleep(500 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error connecting to pod (%v), retrying", netErr) + continue + } + } + } + } + + if err == nil { + err = fission.MakeErrorFromHTTP(resp) + } + + return errors.Wrap(err, "Error specializing function pod") + } + + return errors.Wrap(err, fmt.Sprintf("Error specializing function pod after %v times", maxRetries)) +} diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index 36927df7..590b8c96 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -22,7 +22,6 @@ import ( "fmt" "log" "path/filepath" - "strconv" "time" asv1 "k8s.io/api/autoscaling/v1" @@ -35,7 +34,6 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/environments/fetcher" "github.com/fission/fission/executor/util" ) @@ -44,10 +42,6 @@ const ( DeploymentVersion = "extensions/v1beta1" ) -const ( - envVersion = "ENV_VERSION" -) - func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Environment, deployName string, deployLabels map[string]string, deployNamespace string, firstcreate bool) (*v1beta1.Deployment, error) { @@ -167,29 +161,27 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen gracePeriodSeconds = env.Spec.TerminationGracePeriod } - fetchReq := &fetcher.FetchRequest{ - FetchType: fetcher.FETCH_DEPLOYMENT, - Package: metav1.ObjectMeta{ - Namespace: fn.Spec.Package.PackageRef.Namespace, - Name: fn.Spec.Package.PackageRef.Name, + specializeReq := fission.FunctionSpecializeRequest{ + FetchReq: fission.FunctionFetchRequest{ + FetchType: fission.FETCH_DEPLOYMENT, + Package: metav1.ObjectMeta{ + Namespace: fn.Spec.Package.PackageRef.Namespace, + Name: fn.Spec.Package.PackageRef.Name, + }, + Filename: targetFilename, + Secrets: fn.Spec.Secrets, + ConfigMaps: fn.Spec.ConfigMaps, + KeepArchive: env.Spec.KeepArchive, + }, + LoadReq: fission.FunctionLoadRequest{ + FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), + FunctionName: fn.Spec.Package.FunctionName, + FunctionMetadata: &fn.Metadata, + EnvVersion: env.Spec.Version, }, - Filename: targetFilename, - Secrets: fn.Spec.Secrets, - ConfigMaps: fn.Spec.ConfigMaps, - KeepArchive: env.Spec.KeepArchive, } - loadReq := fission.FunctionLoadRequest{ - FilePath: filepath.Join(deploy.sharedMountPath, targetFilename), - FunctionName: fn.Spec.Package.FunctionName, - FunctionMetadata: &fn.Metadata, - } - - fetchPayload, err := json.Marshal(fetchReq) - if err != nil { - return nil, err - } - loadPayload, err := json.Marshal(loadReq) + specializePayload, err := json.Marshal(specializeReq) if err != nil { return nil, err } @@ -297,8 +289,7 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen }, }, Command: []string{"/fetcher", "-specialize-on-startup", - "-fetch-request", string(fetchPayload), - "-load-request", string(loadPayload), + "-specialize-request", string(specializePayload), "-secret-dir", deploy.sharedSecretPath, "-cfgmap-dir", deploy.sharedCfgMapPath, deploy.sharedMountPath}, @@ -312,12 +303,6 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen }, }, }, - Env: []apiv1.EnvVar{ - { - Name: envVersion, - Value: strconv.Itoa(env.Spec.Version), - }, - }, Resources: fetcherResources, ReadinessProbe: &apiv1.Probe{ InitialDelaySeconds: 1, @@ -325,7 +310,7 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen FailureThreshold: 30, Handler: apiv1.Handler{ HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", + Path: "/readniess-healthz", Port: intstr.IntOrString{ Type: intstr.Int, IntVal: 8000, @@ -334,7 +319,7 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *crd.Function, env *crd.Environmen }, }, LivenessProbe: &apiv1.Probe{ - InitialDelaySeconds: 35, + InitialDelaySeconds: 1, PeriodSeconds: 5, Handler: apiv1.Handler{ HTTPGet: &apiv1.HTTPGetAction{ diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index 0cb8e4e9..3098c69c 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -17,14 +17,10 @@ limitations under the License. package poolmgr import ( - "bytes" - "encoding/json" "fmt" "log" "math/rand" "net" - "net/http" - "net/url" "os" "path/filepath" "strings" @@ -41,7 +37,6 @@ import ( "github.com/fission/fission" "github.com/fission/fission/crd" - "github.com/fission/fission/environments/fetcher" fetcherClient "github.com/fission/fission/environments/fetcher/client" "github.com/fission/fission/executor/fscache" "github.com/fission/fission/executor/util" @@ -282,8 +277,8 @@ func IsIPv6(podIP string) bool { return ip != nil && strings.Contains(podIP, ":") } -func (gp *GenericPool) getFetcherUrl(podIP string) string { - testUrl := os.Getenv("TEST_FETCHER_URL") +func (gp *GenericPool) getSpecializeUrl(podIP string) string { + testUrl := os.Getenv("TEST_SPECIALIZE_URL") if len(testUrl) != 0 { // it takes a second or so for the test service to // become routable once a pod is relabeled. This is @@ -302,26 +297,6 @@ func (gp *GenericPool) getFetcherUrl(podIP string) string { } -func (gp *GenericPool) getSpecializeUrl(podIP string, version int) string { - u := os.Getenv("TEST_SPECIALIZE_URL") - isv6 := IsIPv6(podIP) - var baseUrl string - if len(u) != 0 { - return u - } - if isv6 == false { - baseUrl = fmt.Sprintf("http://%v:8888", podIP) - } else if isv6 == true { // We use bracket if the IP is in IPv6. - baseUrl = fmt.Sprintf("http://[%v]:8888", podIP) - } - - if version == 1 { - return fmt.Sprintf("%v/specialize", baseUrl) - } else { - return fmt.Sprintf("%v/v%v/specialize", baseUrl, version) - } -} - // 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. @@ -338,7 +313,7 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta } // tell fetcher to get the function. - fetcherUrl := gp.getFetcherUrl(podIP) + fetcherUrl := gp.getSpecializeUrl(podIP) log.Printf("[%v] calling fetcher to copy function with fetcher url: %v", metadata.Name, fetcherUrl) fn, err := gp.fissionClient. @@ -356,89 +331,33 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta targetFilename = string(fn.Metadata.UID) } - err = fetcherClient.MakeClient(fetcherUrl).Fetch(&fetcher.FetchRequest{ - FetchType: fetcher.FETCH_DEPLOYMENT, - Package: metav1.ObjectMeta{ - Namespace: fn.Spec.Package.PackageRef.Namespace, - Name: fn.Spec.Package.PackageRef.Name, + specializeReq := fission.FunctionSpecializeRequest{ + FetchReq: fission.FunctionFetchRequest{ + FetchType: fission.FETCH_DEPLOYMENT, + Package: metav1.ObjectMeta{ + Namespace: fn.Spec.Package.PackageRef.Namespace, + Name: fn.Spec.Package.PackageRef.Name, + }, + Filename: targetFilename, + Secrets: fn.Spec.Secrets, + ConfigMaps: fn.Spec.ConfigMaps, + KeepArchive: gp.env.Spec.KeepArchive, + }, + LoadReq: fission.FunctionLoadRequest{ + FilePath: filepath.Join(gp.sharedMountPath, targetFilename), + FunctionName: fn.Spec.Package.FunctionName, + FunctionMetadata: &fn.Metadata, + EnvVersion: gp.env.Spec.Version, }, - Filename: targetFilename, - Secrets: fn.Spec.Secrets, - ConfigMaps: fn.Spec.ConfigMaps, - KeepArchive: gp.env.Spec.KeepArchive, - }) - if err != nil { - return err } - // get function run container to specialize log.Printf("[%v] specializing pod", metadata.Name) - // retry the specialize call a few times in case the env server hasn't come up yet - maxRetries := 20 - - loadReq := fission.FunctionLoadRequest{ - FilePath: filepath.Join(gp.sharedMountPath, targetFilename), - FunctionName: fn.Spec.Package.FunctionName, - FunctionMetadata: &fn.Metadata, - } - - body, err := json.Marshal(loadReq) + err = fetcherClient.MakeClient(fetcherUrl).Specialize(&specializeReq) if err != nil { return err } - for i := 0; i < maxRetries; i++ { - var resp2 *http.Response - if gp.env.Spec.Version == 2 { - specializeUrl := gp.getSpecializeUrl(podIP, 2) - log.Printf("specialize url: %v", specializeUrl) - resp2, err = http.Post(specializeUrl, "application/json", bytes.NewReader(body)) - } else { - specializeUrl := gp.getSpecializeUrl(podIP, 1) - resp2, err = http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{})) - } - - if err == nil && resp2.StatusCode < 300 { - // Success - resp2.Body.Close() - return nil - } - - retry := false - - // Only retry for the specific case of a connection error. - if urlErr, ok := err.(*url.Error); ok { - if netErr, ok := urlErr.Err.(*net.OpError); ok { - if netErr.Op == "dial" { - if i < maxRetries-1 { - retry = true - } - } - } - } - - // Receive response with non-200 http code - if err == nil { - err = fission.MakeErrorFromHTTP(resp2) - } - - // The istio-proxy block all http requests until it's ready - // to serve traffic. Retry if istio feature is enabled. - if gp.useIstio { - retry = true - } - - if retry && i < maxRetries-1 { - time.Sleep(500 * time.Duration(2*i) * time.Millisecond) - log.Printf("Error connecting to pod (%v), retrying", err) - continue - } - - log.Printf("Failed to specialize pod: %v", err) - return err - } - return nil } @@ -593,7 +512,7 @@ func (gp *GenericPool) createPool() error { FailureThreshold: 30, Handler: apiv1.Handler{ HTTPGet: &apiv1.HTTPGetAction{ - Path: "/healthz", + Path: "/readniess-healthz", Port: intstr.IntOrString{ Type: intstr.Int, IntVal: 8000, @@ -602,7 +521,7 @@ func (gp *GenericPool) createPool() error { }, }, LivenessProbe: &apiv1.Probe{ - InitialDelaySeconds: 35, + InitialDelaySeconds: 1, PeriodSeconds: 5, Handler: apiv1.Handler{ HTTPGet: &apiv1.HTTPGetAction{ diff --git a/executor/reaper/reaper.go b/executor/reaper/reaper.go index 54977744..e69174c0 100644 --- a/executor/reaper/reaper.go +++ b/executor/reaper/reaper.go @@ -137,7 +137,6 @@ func cleanupPods(client *kubernetes.Clientset, instanceId string) error { return err } for _, pod := range podList.Items { - log.Printf("Clean pod: %v", pod.ObjectMeta.Name) id, ok := pod.ObjectMeta.Labels[fission.EXECUTOR_INSTANCEID_LABEL] if ok && id != instanceId { log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name) diff --git a/test/tests/test_secret_cfgmap/test_secret_cfgmap.sh b/test/tests/test_secret_cfgmap/test_secret_cfgmap.sh index 92b3b058..2a437f75 100755 --- a/test/tests/test_secret_cfgmap/test_secret_cfgmap.sh +++ b/test/tests/test_secret_cfgmap/test_secret_cfgmap.sh @@ -39,21 +39,22 @@ sed -i "s/{{ FN_CFGMAP }}/${fn_cfgmap}/g" cfgmap.py checkFunctionResponse() { log "Doing an HTTP GET on the function's route" - response=$(curl http://$FISSION_ROUTER/${1}) val=${2} type=${3} log "Checking for valid response" - log ${response} - - if [[ ${response} != ${val} ]] - then - log "test ${type} failed" - cleanup - exit 1 - fi - log "test ${type} passed" + while true; do + log curl http://$FISSION_ROUTER/$1 + response0=$(curl http://$FISSION_ROUTER/$1) + log $response0 | grep -i ${val} + if [[ $? -eq 0 ]]; then + log "test ${type} passed" + break + fi + sleep 1 + done } +export -f checkFunctionResponse # Create a hello world function in nodejs, test it with an http trigger log "Pre-test cleanup" @@ -74,7 +75,7 @@ fission route create --function ${fn_secret} --url /${fn_secret} --method GET log "Waiting for router to catch up" sleep 5 -checkFunctionResponse ${fn_secret} 'TESTVALUE' 'secret' +timeout 60 bash -c "checkFunctionResponse ${fn_secret} 'TESTVALUE' 'secret'" log "Creating function with newdeploy executorType and new secret value" kubectl patch secrets ${fn_secret} -p '{"data":{"TEST_KEY":"TkVXVkFMCg=="}}' -n default @@ -86,7 +87,7 @@ fission route create --function ${fn_secret}-1 --url /${fn_secret}-1 --method GE log "Waiting for router catch up" sleep 5 -checkFunctionResponse ${fn_secret}-1 'NEWVAL' 'secret' +timeout 60 bash -c "checkFunctionResponse ${fn_secret}-1 'NEWVAL' 'secret'" log "Creating configmap" kubectl create configmap ${fn_cfgmap} --from-literal=TEST_KEY="TESTVALUE" -n default @@ -100,7 +101,7 @@ fission route create --function ${fn_cfgmap} --url /${fn_cfgmap} --method GET log "Waiting for router to catch up" sleep 5 -checkFunctionResponse ${fn_cfgmap} 'TESTVALUE' 'configmap' +timeout 60 bash -c "checkFunctionResponse ${fn_cfgmap} 'TESTVALUE' 'configmap'" log "Creating function with newdeploy executorType and new configmap value" kubectl patch configmap ${fn_cfgmap} -p '{"data":{"TEST_KEY":"NEWVAL"}}' -n default @@ -112,7 +113,7 @@ fission route create --function ${fn_cfgmap}-1 --url /${fn_cfgmap}-1 --method GE log "Waiting for router catch up" sleep 5 -checkFunctionResponse ${fn_cfgmap}-1 'NEWVAL' 'configmap' +timeout 60 bash -c "checkFunctionResponse ${fn_cfgmap}-1 'NEWVAL' 'configmap'" log "testing creating a function without a secret or configmap" fission function create --name ${fn} --env python --code empty.py @@ -124,13 +125,4 @@ log "Waiting for router to catch up" sleep 5 log "HTTP GET on the function's route" -resnormal=$(curl http://${FISSION_ROUTER}/${fn}) -if [ ${resnormal} != "yes" ] -then - log "test empty failed" - cleanup - exit 1 -fi -log "test empty passed" - -log "All done." +timeout 60 bash -c "checkFunctionResponse ${fn} 'yes' 'configmap'" diff --git a/types.go b/types.go index e51c0dfb..8ad32a1d 100644 --- a/types.go +++ b/types.go @@ -73,6 +73,24 @@ type ( // talk to environments. // type ( + FetchRequestType int + + FunctionSpecializeRequest struct { + FetchReq FunctionFetchRequest + LoadReq FunctionLoadRequest + } + + FunctionFetchRequest struct { + FetchType FetchRequestType `json:"fetchType"` + Package metav1.ObjectMeta `json:"package"` + Url string `json:"url"` + StorageSvcUrl string `json:"storagesvcurl"` + Filename string `json:"filename"` + Secrets []SecretReference `json:"secretList"` + ConfigMaps []ConfigMapReference `json:"configMapList"` + KeepArchive bool `json:"keeparchive"` + } + FunctionLoadRequest struct { // FilePath is an absolute filesystem path to the // function. What exactly is stored here is @@ -91,7 +109,30 @@ type ( // Metatdata FunctionMetadata *metav1.ObjectMeta + + EnvVersion int `json:"envVersion"` } + + // ArchiveUploadRequest send from builder manager describes which + // deployment package should be upload to storage service. + ArchiveUploadRequest struct { + Filename string `json:"filename"` + StorageSvcUrl string `json:"storagesvcurl"` + ArchivePackage bool `json:"archivepackage"` + } + + // ArchiveUploadResponse defines the download url of an archive and + // its checksum. + ArchiveUploadResponse struct { + ArchiveDownloadUrl string `json:"archiveDownloadUrl"` + Checksum Checksum `json:"checksum"` + } +) + +const ( + FETCH_SOURCE = iota + FETCH_DEPLOYMENT + FETCH_URL // remove this? ) const EXECUTOR_INSTANCEID_LABEL = fv1.EXECUTOR_INSTANCEID_LABEL