add ota update RESTful API

cmd/yurthub/app/start.go

@@ -195,7 +195,7 @@ func Run(cfg *config.YurtHubConfiguration, stopCh <-chan struct{}) error {
 	cfg.YurtSharedFactory.Start(stopCh)
 
 	klog.Infof("%d. new %s server and begin to serve, proxy server: %s, secure proxy server: %s, hub server: %s", trace, projectinfo.GetHubName(), cfg.YurtHubProxyServerAddr, cfg.YurtHubProxyServerSecureAddr, cfg.YurtHubServerAddr)
-	s, err := server.NewYurtHubServer(cfg, certManager, yurtProxyHandler, restConfigMgr)
+	s, err := server.NewYurtHubServer(cfg, certManager, yurtProxyHandler, restConfigMgr, healthChecker)
 	if err != nil {
 		return fmt.Errorf("could not create hub server, %w", err)
 	}

pkg/yurthub/otaupdate/ota.go

@@ -0,0 +1,177 @@
+/*
+Copyright 2022 The OpenYurt Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package otaupdate
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+
+	"github.com/gorilla/mux"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	runtimescheme "k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/klog/v2"
+
+	"github.com/openyurtio/openyurt/pkg/controller/daemonpodupdater"
+	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
+)
+
+// GetPods return pod list
+func GetPods(clientset kubernetes.Interface, nodeName string, checker healthchecker.HealthChecker,
+	servers []*url.URL) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Pre-check if edge yurthub node is connected to the cloud
+		if !isEdgeCloudConnected(checker, servers) {
+			klog.Errorf("Get pod list is not allowed when edge is disconnected to cloud")
+			WriteErr(w, "Get pod list is not allowed when edge is disconnected to cloud", http.StatusForbidden)
+			return
+		}
+
+		podList, err := clientset.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{
+			FieldSelector: "spec.nodeName=" + nodeName,
+		})
+		if err != nil {
+			klog.Errorf("Get pod list failed, %v", err)
+			WriteErr(w, "Get pod list failed", http.StatusInternalServerError)
+			return
+		}
+		klog.V(5).Infof("Got pod list: %v", podList)
+
+		// Successfully get pod list, response 200
+		data, err := encodePodList(podList)
+		if err != nil {
+			klog.Errorf("Encode pod list failed, %v", err)
+			WriteErr(w, "Encode pod list failed", http.StatusInternalServerError)
+		}
+		WriteJSONResponse(w, data)
+	})
+}
+
+// UpdatePod update a specifc pod(namespace/podname) to the latest version
+func UpdatePod(clientset kubernetes.Interface, nodeName string, checker healthchecker.HealthChecker,
+	servers []*url.URL) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Pre-check if edge yurthub node is connected to the cloud
+		if !isEdgeCloudConnected(checker, servers) {
+			klog.Errorf("Apply update is not allowed when edge is disconnected to cloud")
+			WriteErr(w, "Apply update is not allowed when edge is disconnected to cloud", http.StatusForbidden)
+			return
+		}
+
+		params := mux.Vars(r)
+		namespace := params["ns"]
+		podName := params["podname"]
+
+		err, ok := applyUpdate(clientset, namespace, podName, nodeName)
+		// Pod update failed with error
+		if err != nil {
+			WriteErr(w, "Apply update failed", http.StatusInternalServerError)
+			return
+		}
+		// Pod update is not allowed
+		if !ok {
+			WriteErr(w, "Pod is not-updatable", http.StatusForbidden)
+			return
+		}
+
+		// Successfully apply update, response 200
+		WriteJSONResponse(w, []byte(fmt.Sprintf("Start updating pod %q/%q", namespace, podName)))
+	})
+}
+
+// applyUpdate execute pod update process by deleting pod under OnDelete update strategy
+func applyUpdate(clientset kubernetes.Interface, namespace, podName, nodeName string) (error, bool) {
+	pod, err := clientset.CoreV1().Pods(namespace).Get(context.TODO(), podName, metav1.GetOptions{})
+	if err != nil {
+		klog.Errorf("Get pod %v/%v failed, %v", namespace, podName, err)
+		return err, false
+	}
+
+	// Pod will not be updated when it's being deleted
+	if pod.DeletionTimestamp != nil {
+		klog.Infof("Pod %v/%v is deleting, can not be updated", namespace, podName)
+		return nil, false
+	}
+
+	// Pod will not be updated when it's not running on the current node
+	if pod.Spec.NodeName != nodeName {
+		klog.Infof("Pod: %v/%v is running on %v, can not be updated", namespace, podName, pod.Spec.NodeName)
+		return nil, false
+	}
+
+	// Pod will not be updated without pod condition PodNeedUpgrade=true
+	if !daemonpodupdater.IsPodUpdatable(pod) {
+		klog.Infof("Pod: %v/%v is not updatable", namespace, podName)
+		return nil, false
+	}
+
+	klog.V(5).Infof("Pod: %v/%v is updatable", namespace, podName)
+	err = clientset.CoreV1().Pods(namespace).Delete(context.TODO(), podName, metav1.DeleteOptions{})
+	if err != nil {
+		klog.Errorf("Update pod %v/%v failed when delete, %v", namespace, podName, err)
+		return err, false
+	}
+
+	klog.Infof("Start updating pod: %q/%q", namespace, podName)
+	return nil, true
+}
+
+// encodePodList returns the encoded PodList
+func encodePodList(podList *corev1.PodList) ([]byte, error) {
+	codec := scheme.Codecs.LegacyCodec(runtimescheme.GroupVersion{Group: corev1.GroupName, Version: "v1"})
+	return runtime.Encode(codec, podList)
+}
+
+// isEdgeCloudConnected will check if edge is disconnected to cloud. If there is any remote server is healthy, it is
+// regarded as connected. Otherwise, it is regarded as disconnected and return false.
+func isEdgeCloudConnected(checker healthchecker.HealthChecker, remoteServers []*url.URL) bool {
+	for _, server := range remoteServers {
+		if checker.IsHealthy(server) {
+			return true
+		}
+	}
+	return false
+}
+
+// WriteErr writes the http status and the error string on the response
+func WriteErr(w http.ResponseWriter, errReason string, httpStatus int) {
+	w.WriteHeader(httpStatus)
+	n := len([]byte(errReason))
+	nw, e := w.Write([]byte(errReason))
+	if e != nil || nw != n {
+		klog.Errorf("Write resp for request, expect %d bytes but write %d bytes with error, %v", n, nw, e)
+	}
+}
+
+// Derived from kubelet writeJSONResponse
+func WriteJSONResponse(w http.ResponseWriter, data []byte) {
+	if data == nil {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	n, err := w.Write(data)
+	if err != nil || n != len(data) {
+		klog.Errorf("Write resp for request, expect %d bytes but write %d bytes with error, %v", len(data), n, err)
+	}
+}

pkg/yurthub/otaupdate/ota_test.go

@@ -0,0 +1,157 @@
+/*
+Copyright 2022 The OpenYurt Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package otaupdate
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	"github.com/openyurtio/openyurt/pkg/controller/daemonpodupdater"
+	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
+)
+
+var (
+	healthyServers = []*url.URL{
+		{Host: "127.0.0.1:18080"},
+	}
+
+	unHealthyServers = []*url.URL{
+		{Host: "127.0.0.1:18081"},
+	}
+
+	healthyFakeChecker = healthchecker.NewFakeChecker(true, map[string]int{
+		"http://127.0.0.1:8080": 1,
+	})
+
+	unHealthyFakeChecker = healthchecker.NewFakeChecker(false, map[string]int{
+		"http://127.0.0.1:8081": 1,
+	})
+)
+
+func newPod(podName string) *corev1.Pod {
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{APIVersion: "v1"},
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: podName,
+			Namespace:    metav1.NamespaceDefault,
+		},
+		Status: corev1.PodStatus{
+			Conditions: []corev1.PodCondition{},
+		},
+	}
+	pod.Name = podName
+	return pod
+}
+
+func newPodWithCondition(podName string, ready corev1.ConditionStatus) *corev1.Pod {
+	pod := newPod(podName)
+	SetPodUpgradeCondition(pod, ready)
+
+	return pod
+}
+
+func SetPodUpgradeCondition(pod *corev1.Pod, ready corev1.ConditionStatus) {
+	cond := corev1.PodCondition{
+		Type:   daemonpodupdater.PodNeedUpgrade,
+		Status: ready,
+	}
+	pod.Status.Conditions = append(pod.Status.Conditions, cond)
+}
+
+func TestGetPods(t *testing.T) {
+	updatablePod := newPodWithCondition("updatablePod", corev1.ConditionTrue)
+	notUpdatablePod := newPodWithCondition("notUpdatablePod", corev1.ConditionFalse)
+	normalPod := newPod("normalPod")
+
+	clientset := fake.NewSimpleClientset(updatablePod, notUpdatablePod, normalPod)
+
+	req, err := http.NewRequest("GET", "/openyurt.io/v1/pods", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr := httptest.NewRecorder()
+
+	GetPods(clientset, "", healthyFakeChecker, healthyServers).ServeHTTP(rr, req)
+
+	expectedCode := http.StatusOK
+	assert.Equal(t, expectedCode, rr.Code)
+
+	// Cloud-Edge network disconnected
+	rr = httptest.NewRecorder()
+	GetPods(clientset, "", unHealthyFakeChecker, unHealthyServers).ServeHTTP(rr, req)
+	assert.Equal(t, http.StatusForbidden, rr.Code)
+}
+
+func TestUpdatePod(t *testing.T) {
+	tests := []struct {
+		reqURL       string
+		pod          *corev1.Pod
+		podName      string
+		expectedCode int
+		expectedData string
+	}{
+		{
+			reqURL:       "/openyurt.io/v1/namespaces/default/pods/updatablePod/update",
+			podName:      "updatablePod",
+			pod:          newPodWithCondition("updatablePod", corev1.ConditionTrue),
+			expectedCode: http.StatusOK,
+			expectedData: "Start updating pod \"default\"/\"updatablePod\"",
+		},
+		{
+			reqURL:       "/openyurt.io/v1/namespaces/default/pods/notUpdatablePod/update",
+			podName:      "notUpdatablePod",
+			pod:          newPodWithCondition("notUpdatablePod", corev1.ConditionFalse),
+			expectedCode: http.StatusForbidden,
+			expectedData: "Pod is not-updatable",
+		},
+		{
+			reqURL:       "/openyurt.io/v1/namespaces/default/pods/wrongName/update",
+			podName:      "wrongName",
+			pod:          newPodWithCondition("trueName", corev1.ConditionFalse),
+			expectedCode: http.StatusInternalServerError,
+			expectedData: "Apply update failed",
+		},
+	}
+	for _, test := range tests {
+		clientset := fake.NewSimpleClientset(test.pod)
+
+		req, err := http.NewRequest("POST", test.reqURL, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		vars := map[string]string{
+			"ns":      "default",
+			"podname": test.podName,
+		}
+		req = mux.SetURLVars(req, vars)
+		rr := httptest.NewRecorder()
+
+		UpdatePod(clientset, "", healthyFakeChecker, healthyServers).ServeHTTP(rr, req)
+
+		assert.Equal(t, test.expectedCode, rr.Code)
+		assert.Equal(t, test.expectedData, rr.Body.String())
+	}
+
+}

pkg/yurthub/server/server.go

@@ -37,7 +37,9 @@ import (
 	"github.com/openyurtio/openyurt/pkg/util/certmanager"
 	certfactory "github.com/openyurtio/openyurt/pkg/util/certmanager/factory"
 	"github.com/openyurtio/openyurt/pkg/yurthub/certificate/interfaces"
+	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/rest"
+	ota "github.com/openyurtio/openyurt/pkg/yurthub/otaupdate"
 )
 
 // Server is an interface for providing http service for yurthub
@@ -60,15 +62,16 @@ type yurtHubServer struct {
 func NewYurtHubServer(cfg *config.YurtHubConfiguration,
 	certificateMgr interfaces.YurtCertificateManager,
 	proxyHandler http.Handler,
-	rest *rest.RestConfigManager) (Server, error) {
+	rest *rest.RestConfigManager,
+	checker healthchecker.HealthChecker) (Server, error) {
 	hubMux := mux.NewRouter()
-	registerHandlers(hubMux, cfg, certificateMgr)
 	restCfg := rest.GetRestConfig(false)
 	clientSet, err := kubernetes.NewForConfig(restCfg)
 	if err != nil {
 		klog.Errorf("cannot create the client set: %v", err)
 		return nil, err
 	}
+	registerHandlers(hubMux, cfg, certificateMgr, clientSet, checker)
 	hubServer := &http.Server{
 		Addr:           cfg.YurtHubServerAddr,
 		Handler:        hubMux,
@@ -156,7 +159,8 @@ func (s *yurtHubServer) Run() {
 }
 
 // registerHandler registers handlers for yurtHubServer, and yurtHubServer can handle requests like profiling, healthz, update token.
-func registerHandlers(c *mux.Router, cfg *config.YurtHubConfiguration, certificateMgr interfaces.YurtCertificateManager) {
+func registerHandlers(c *mux.Router, cfg *config.YurtHubConfiguration, certificateMgr interfaces.YurtCertificateManager,
+	clientset *kubernetes.Clientset, checker healthchecker.HealthChecker) {
 	// register handlers for update join token
 	c.Handle("/v1/token", updateTokenHandler(certificateMgr)).Methods("POST", "PUT")
 
@@ -170,6 +174,11 @@ func registerHandlers(c *mux.Router, cfg *config.YurtHubConfiguration, certifica
 
 	// register handler for metrics
 	c.Handle("/metrics", promhttp.Handler())
+
+	// register handler for ota upgrade
+	c.Handle("/pods", ota.GetPods(clientset, cfg.NodeName, checker, cfg.RemoteServers)).Methods("GET")
+	c.Handle("/openyurt.io/v1/namespaces/{ns}/pods/{podname}/upgrade",
+		ota.UpdatePod(clientset, cfg.NodeName, checker, cfg.RemoteServers)).Methods("POST")
 }
 
 // healthz returns ok for healthz request