diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 23a91051..be290066 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -215,25 +215,28 @@ func (gp *GenericPool) labelsForMetadata(metadata *fission.Metadata) map[string] } } +func (gp *GenericPool) scheduleDeletePod(name string) { + go func() { + // The sleep allows debugging or collecting logs from the pod before it's + // cleaned up. (We need a better solutions for both those things; log + // aggregation and storage will help.) + log.Printf("Error in pod '%v', scheduling cleanup", name) + time.Sleep(5 * time.Minute) + gp.kubernetesClient.Core().Pods(gp.namespace).Delete(name, nil) + }() +} + // 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) (*v1.Pod, error) { - newLabels := gp.labelsForMetadata(metadata) - - log.Printf("[%v] Choosing pod from pool", metadata) - pod, err := gp.choosePod(newLabels) - if err != nil { - return nil, err - } - +func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *fission.Metadata) error { // 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") + return errors.New("Pod has no IP") } - // tell fetcher to get the function + // tell fetcher to get the function. fetcherUrl := fmt.Sprintf("http://%v:8000/", podIP) functionUrl := fmt.Sprintf("%v/v1/functions/%v?uid=%v&raw=1", gp.controllerUrl, metadata.Name, metadata.Uid) @@ -243,11 +246,11 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error resp, err := http.Post(fetcherUrl, "application/json", bytes.NewReader([]byte(fetcherRequest))) if err != nil { // TODO we should retry this call in case fetcher hasn't come up yet - return nil, err + return err } defer resp.Body.Close() if resp.StatusCode != 200 { - return nil, errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status)) + return errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status)) } // get function run container to specialize @@ -259,8 +262,9 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error for i := 0; i < maxRetries; i++ { resp2, err := http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{})) if err == nil && resp2.StatusCode < 300 { + // Success resp2.Body.Close() - return pod, nil + return nil } // Only retry for the specific case of a connection error. @@ -280,10 +284,10 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error err = fission.MakeErrorFromHTTP(resp2) } log.Printf("Failed to specialize pod: %v", err) - return nil, err + return err } - return pod, nil + return nil } // A pool is a deployment of generic containers for an env. This @@ -407,10 +411,19 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Ser } func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { - pod, err := gp.specializePod(m) + + log.Printf("[%v] Choosing pod from pool", m) + newLabels := gp.labelsForMetadata(m) + pod, err := gp.choosePod(newLabels) if err != nil { return nil, err } + + err = gp.specializePod(pod, m) + if err != nil { + gp.scheduleDeletePod(pod.ObjectMeta.Name) + return nil, err + } log.Printf("Specialized pod: %v", pod.ObjectMeta.Name) var svcHost string @@ -423,9 +436,11 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { labels := gp.labelsForMetadata(m) svc, err := gp.createSvc(svcName, labels) if err != nil { + gp.scheduleDeletePod(pod.ObjectMeta.Name) return nil, err } if svc.ObjectMeta.Name != svcName { + gp.scheduleDeletePod(pod.ObjectMeta.Name) return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name)) }