Initialize services for hyper pod

Author: feiskyer

pkg/kubelet/hyper/hyper.go

@@ -35,10 +35,12 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	"k8s.io/kubernetes/pkg/client/record"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/credentialprovider"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	"k8s.io/kubernetes/pkg/kubelet/prober"
+	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/probe"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util"
@@ -66,6 +68,7 @@ type runtime struct {
 	readinessManager    *kubecontainer.ReadinessManager
 	volumeGetter        volumeGetter
 	hyperClient         *HyperClient
+	kubeClient          client.Interface
 	imagePuller         kubecontainer.ImagePuller
 }
 
@@ -81,7 +84,8 @@ func New(generator kubecontainer.RunContainerOptionsGenerator,
 	networkPlugin network.NetworkPlugin,
 	containerRefManager *kubecontainer.RefManager,
 	readinessManager *kubecontainer.ReadinessManager,
-	volumeGetter volumeGetter) (kubecontainer.Runtime, error) {
+	volumeGetter volumeGetter,
+	kubeClient client.Interface) (kubecontainer.Runtime, error) {
 
 	// check hyper has already installed
 	hyperBinAbsPath, err := exec.LookPath(hyperBinName)
@@ -100,6 +104,7 @@ func New(generator kubecontainer.RunContainerOptionsGenerator,
 		readinessManager:    readinessManager,
 		volumeGetter:        volumeGetter,
 		hyperClient:         NewHyperClient(),
+		kubeClient:          kubeClient,
 	}
 	hyper.prober = prober.New(hyper, readinessManager, containerRefManager, recorder)
 	hyper.imagePuller = kubecontainer.NewImagePuller(recorder, hyper)
@@ -309,6 +314,40 @@ func (r *runtime) GetPods(all bool) ([]*kubecontainer.Pod, error) {
 	return kubepods, nil
 }
 
+func (r *runtime) buildHyperPodServices(pod *api.Pod) []HyperService {
+	items, err := r.kubeClient.Services(pod.Namespace).List(labels.Everything())
+	if err != nil {
+		glog.Warningf("Get services failed: %v", err)
+		return nil
+	}
+
+	var services []HyperService
+	for _, svc := range items.Items {
+		hyperService := HyperService{
+			ServiceIP: svc.Spec.ClusterIP,
+		}
+		endpoints, _ := r.kubeClient.Endpoints(pod.Namespace).Get(svc.Name)
+		for _, svcPort := range svc.Spec.Ports {
+			hyperService.ServicePort = svcPort.Port
+			for _, ep := range endpoints.Subsets {
+				for _, epPort := range ep.Ports {
+					if svcPort.Name == "" || svcPort.Name == epPort.Name {
+						for _, eh := range ep.Addresses {
+							hyperService.Hosts = append(hyperService.Hosts, HyperServiceBackend{
+								HostIP:   eh.IP,
+								HostPort: epPort.Port,
+							})
+						}
+					}
+				}
+			}
+			services = append(services, hyperService)
+		}
+	}
+
+	return services
+}
+
 func (r *runtime) buildHyperPod(pod *api.Pod, pullSecrets []api.Secret) ([]byte, error) {
 	// check and pull image
 	for _, c := range pod.Spec.Containers {
@@ -335,6 +374,18 @@ func (r *runtime) buildHyperPod(pod *api.Pod, pullSecrets []api.Secret) ([]byte,
 	}
 	specMap[KEY_VOLUMES] = volumes
 
+	services := r.buildHyperPodServices(pod)
+	if services == nil {
+		// Just for fake
+		services = []HyperService{
+			{
+				ServiceIP:   "127.0.0.2",
+				ServicePort: 65534,
+			},
+		}
+	}
+	specMap["services"] = services
+
 	// build hyper containers spec
 	var containers []map[string]interface{}
 	for _, container := range pod.Spec.Containers {

pkg/kubelet/kubelet.go

@@ -437,7 +437,9 @@ func NewMainKubelet(
 			klet.networkPlugin,
 			containerRefManager,
 			readinessManager,
-			klet.volumeManager)
+			klet.volumeManager,
+			klet.kubeClient,
+		)
 		if err != nil {
 			return nil, err
 		}

pkg/proxy/haproxy/proxier.go

@@ -353,15 +353,6 @@ func (proxier *Proxier) syncProxyRules() {
 			continue
 		}
 
-		// Get hyper services
-		hyperServices, err := proxier.hyperClient.ListServices(podInfo.PodID)
-		if err != nil {
-			glog.Warningf("Can not get hyper service for pod %s: %v", podInfo.PodName, err)
-			continue
-		}
-
-		glog.V(4).Infof("Hyper services of pod %s is %v", podInfo.PodName, hyperServices)
-
 		// Build rules in same namespace
 		consumedServices := make([]hyper.HyperService, 0, 1)
 		for _, svcInfo := range proxier.serviceMap {
@@ -391,7 +382,7 @@ func (proxier *Proxier) syncProxyRules() {
 		glog.V(4).Infof("Services of pod %s should consumed: %v", podInfo.PodName, consumedServices)
 
 		// update existing services
-		err = proxier.hyperClient.UpdateServices(podInfo.PodID, hyperServices)
+		err = proxier.hyperClient.UpdateServices(podInfo.PodID, consumedServices)
 		if err != nil {
 			glog.Warningf("Updating service for hyper pod %s failed: %v", podInfo.PodName, err)
 		}