Poolmgr -- add retries to specialize call
If the pool is new or very busy, we may call the chosen container's specialize endpoint before it's actually up -- handle the case by retrying a few times. Also namespace-qualify service hostname (since router and functions run in different namespaces).
This commit is contained in:
+39
-12
@@ -22,7 +22,9 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"math/rand"
|
"math/rand"
|
||||||
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/url"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"k8s.io/client-go/1.4/kubernetes"
|
"k8s.io/client-go/1.4/kubernetes"
|
||||||
@@ -30,6 +32,7 @@ import (
|
|||||||
"k8s.io/client-go/1.4/pkg/api/v1"
|
"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/apis/extensions/v1beta1"
|
||||||
"k8s.io/client-go/1.4/pkg/labels"
|
"k8s.io/client-go/1.4/pkg/labels"
|
||||||
|
"k8s.io/client-go/1.4/pkg/util/intstr"
|
||||||
|
|
||||||
"github.com/platform9/fission"
|
"github.com/platform9/fission"
|
||||||
)
|
)
|
||||||
@@ -139,14 +142,15 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
readyPods := make([]v1.Pod, len(podList.Items))
|
readyPods := make([]*v1.Pod, 0, len(podList.Items))
|
||||||
for _, pod := range podList.Items {
|
for _, pod := range podList.Items {
|
||||||
podReady := true
|
podReady := true
|
||||||
for _, cs := range pod.Status.ContainerStatuses {
|
for _, cs := range pod.Status.ContainerStatuses {
|
||||||
podReady = podReady && cs.Ready
|
podReady = podReady && cs.Ready
|
||||||
}
|
}
|
||||||
if podReady {
|
if podReady {
|
||||||
readyPods = append(readyPods, pod)
|
log.Printf("pod %v is ready", pod.ObjectMeta.Name)
|
||||||
|
readyPods = append(readyPods, &pod)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
log.Printf("[%v] found %v ready pods of %v total",
|
log.Printf("[%v] found %v ready pods of %v total",
|
||||||
@@ -170,13 +174,14 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
|
|||||||
// modified, this should fail; in that case just
|
// modified, this should fail; in that case just
|
||||||
// retry.
|
// retry.
|
||||||
chosenPod.ObjectMeta.Labels = newLabels
|
chosenPod.ObjectMeta.Labels = newLabels
|
||||||
_, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(&chosenPod)
|
log.Printf("relabeling pod: [%v]", chosenPod.ObjectMeta.Name)
|
||||||
|
_, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(chosenPod)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("failed to relabel pod: %v", err)
|
log.Printf("failed to relabel pod [%v]: %v", chosenPod.ObjectMeta.Name, err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
log.Printf("Chosen pod: %v (in %v)", chosenPod.ObjectMeta.Name, time.Now().Sub(startTime))
|
log.Printf("Chosen pod: %v (in %v)", chosenPod.ObjectMeta.Name, time.Now().Sub(startTime))
|
||||||
return &chosenPod, nil
|
return chosenPod, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -224,11 +229,28 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error
|
|||||||
// get function run container to specialize
|
// get function run container to specialize
|
||||||
log.Printf("[%v] specializing pod", metadata)
|
log.Printf("[%v] specializing pod", metadata)
|
||||||
specializeUrl := fmt.Sprintf("http://%v:8888/specialize", podIP)
|
specializeUrl := fmt.Sprintf("http://%v:8888/specialize", podIP)
|
||||||
resp2, err := http.Post(specializeUrl, "", bytes.NewReader([]byte{}))
|
|
||||||
if err != nil {
|
// retry the specialize call a few times in case the env server hasn't come up yet
|
||||||
return nil, err
|
maxRetries := 20
|
||||||
|
for i := 0; i < maxRetries; i++ {
|
||||||
|
resp2, err := http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{}))
|
||||||
|
if err != nil {
|
||||||
|
if urlErr, ok := err.(*url.Error); ok {
|
||||||
|
if netErr, ok := urlErr.Err.(*net.OpError); ok {
|
||||||
|
if netErr.Op == "dial" { // && netErr.Err == syscall.ECONNREFUSED
|
||||||
|
log.Printf("Error connecting to pod (%v)", netErr)
|
||||||
|
if i < maxRetries-1 {
|
||||||
|
time.Sleep(500 * time.Duration(2*i) * time.Millisecond)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
resp2.Body.Close()
|
||||||
}
|
}
|
||||||
resp2.Body.Close()
|
|
||||||
return pod, nil
|
return pod, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -338,8 +360,9 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Ser
|
|||||||
Type: v1.ServiceTypeClusterIP,
|
Type: v1.ServiceTypeClusterIP,
|
||||||
Ports: []v1.ServicePort{
|
Ports: []v1.ServicePort{
|
||||||
v1.ServicePort{
|
v1.ServicePort{
|
||||||
Protocol: v1.ProtocolTCP,
|
Protocol: v1.ProtocolTCP,
|
||||||
Port: 8888,
|
Port: 80,
|
||||||
|
TargetPort: intstr.FromInt(8888),
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
Selector: labels,
|
Selector: labels,
|
||||||
@@ -370,10 +393,14 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
|
|||||||
return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name))
|
return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// the fission router isn't in the same namespace, so return a
|
||||||
|
// namespace-qualified hostname
|
||||||
|
svcHost := fmt.Sprintf("%v.%v", svcName, gp.namespace)
|
||||||
|
|
||||||
fsvc := &funcSvc{
|
fsvc := &funcSvc{
|
||||||
function: m,
|
function: m,
|
||||||
environment: gp.env,
|
environment: gp.env,
|
||||||
serviceName: svcName,
|
serviceName: svcHost,
|
||||||
ctime: time.Now(),
|
ctime: time.Now(),
|
||||||
atime: time.Now(),
|
atime: time.Now(),
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user