retrive pods and logs by gateway

Signed-off-by: Roland.Ma <rolandma@kubesphere.io>
2021-09-15 07:55:58 +00:00
parent 81c19701ef
commit e3a14ca299
5 changed files with 247 additions and 13 deletions
--- a/pkg/apiserver/apiserver.go
+++ b/pkg/apiserver/apiserver.go
@@ -273,7 +273,7 @@ func (s *APIServer) installKubeSphereAPIs() {
 	urlruntime.Must(notificationkapisv2beta1.AddToContainer(s.container, s.InformerFactory, s.KubernetesClient.Kubernetes(),
 		s.KubernetesClient.KubeSphere()))
 	urlruntime.Must(notificationkapisv2beta2.AddToContainer(s.container, s.Config.NotificationOptions))
-	urlruntime.Must(gatewayv1alpha1.AddToContainer(s.container, s.Config.GatewayOptions, s.RuntimeCache, s.RuntimeClient))
+	urlruntime.Must(gatewayv1alpha1.AddToContainer(s.container, s.Config.GatewayOptions, s.RuntimeCache, s.RuntimeClient, s.InformerFactory, s.KubernetesClient.Kubernetes()))
 }

 func (s *APIServer) Run(ctx context.Context) (err error) {
--- a/pkg/kapis/gateway/v1alpha1/handler.go
+++ b/pkg/kapis/gateway/v1alpha1/handler.go
@@ -17,30 +17,44 @@ limitations under the License.
 package v1alpha1

 import (
+	"context"
+	"fmt"
+
 	"github.com/emicklei/go-restful"
-	"kubesphere.io/api/gateway/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apiserver/pkg/util/flushwriter"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"

+	"kubesphere.io/api/gateway/v1alpha1"
+
 	"kubesphere.io/kubesphere/pkg/api"
 	"kubesphere.io/kubesphere/pkg/apiserver/query"
+	"kubesphere.io/kubesphere/pkg/informers"
 	operator "kubesphere.io/kubesphere/pkg/models/gateway"
+
 	servererr "kubesphere.io/kubesphere/pkg/server/errors"
 	"kubesphere.io/kubesphere/pkg/simple/client/gateway"
+	conversionsv1 "kubesphere.io/kubesphere/pkg/utils/conversions/core/v1"
 )

 type handler struct {
 	options *gateway.Options
 	gw      operator.GatewayOperator
+	factory informers.InformerFactory
 }

 //newHandler create an instance of the handler
-func newHandler(options *gateway.Options, cache cache.Cache, client client.Client) *handler {
+func newHandler(options *gateway.Options, cache cache.Cache, client client.Client, factory informers.InformerFactory, k8sClient kubernetes.Interface) *handler {
+	conversionsv1.RegisterConversions(scheme.Scheme)
 	// Do not register Gateway scheme globally. Which will cause conflict in ks-controller-manager.
 	v1alpha1.AddToScheme(client.Scheme())
 	return &handler{
 		options: options,
-		gw:      operator.NewGatewayOperator(client, cache, options),
+		factory: factory,
+		gw:      operator.NewGatewayOperator(client, cache, options, factory, k8sClient),
 	}
 }

@@ -126,3 +140,36 @@ func (h *handler) List(request *restful.Request, response *restful.Response) {

 	response.WriteEntity(result)
 }
+
+func (h *handler) ListPods(request *restful.Request, response *restful.Response) {
+	queryParam := query.ParseQueryParameter(request)
+	ns := request.PathParameter("namespace")
+
+	result, err := h.gw.GetPods(ns, queryParam)
+	if err != nil {
+		api.HandleError(response, request, err)
+		return
+	}
+
+	response.WriteEntity(result)
+}
+
+func (h *handler) PodLog(request *restful.Request, response *restful.Response) {
+
+	podNamespace := request.PathParameter("namespace")
+	podID := request.PathParameter("pod")
+
+	query := request.Request.URL.Query()
+	logOptions := &corev1.PodLogOptions{}
+	if err := scheme.ParameterCodec.DecodeParameters(query, corev1.SchemeGroupVersion, logOptions); err != nil {
+		api.HandleError(response, request, fmt.Errorf("unable to decode query"))
+		return
+	}
+
+	fw := flushwriter.Wrap(response.ResponseWriter)
+	err := h.gw.GetPodLogs(context.TODO(), podNamespace, podID, logOptions, fw)
+	if err != nil {
+		api.HandleError(response, request, err)
+		return
+	}
+}
--- a/pkg/kapis/gateway/v1alpha1/register.go
+++ b/pkg/kapis/gateway/v1alpha1/register.go
@@ -22,6 +22,7 @@ import (
 	"github.com/emicklei/go-restful"
 	restfulspec "github.com/emicklei/go-restful-openapi"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
 	"kubesphere.io/api/gateway/v1alpha1"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -29,16 +30,17 @@ import (
 	"kubesphere.io/kubesphere/pkg/api"
 	"kubesphere.io/kubesphere/pkg/apiserver/runtime"
 	"kubesphere.io/kubesphere/pkg/constants"
+	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/server/errors"
 	"kubesphere.io/kubesphere/pkg/simple/client/gateway"
 )

 var GroupVersion = schema.GroupVersion{Group: "gateway.kubesphere.io", Version: "v1alpha1"}

-func AddToContainer(container *restful.Container, options *gateway.Options, cache cache.Cache, client client.Client) error {
+func AddToContainer(container *restful.Container, options *gateway.Options, cache cache.Cache, client client.Client, factory informers.InformerFactory, k8sClient kubernetes.Interface) error {
 	ws := runtime.NewWebService(GroupVersion)

-	handler := newHandler(options, cache, client)
+	handler := newHandler(options, cache, client, factory, k8sClient)

 	// register gateway apis
 	ws.Route(ws.POST("/namespaces/{namespace}/gateways").
@@ -85,6 +87,21 @@ func AddToContainer(container *restful.Container, options *gateway.Options, cach
 		Returns(http.StatusOK, api.StatusOK, v1alpha1.Gateway{}).
 		Metadata(restfulspec.KeyOpenAPITags, []string{constants.GatewayTag}))

+	ws.Route(ws.GET("/namespaces/{namespace}/gateways/{gateway}/pods").
+		To(handler.ListPods).
+		Doc("Retrieve gateways workload pods.").
+		Param(ws.PathParameter("namespace", "the watching namespace of the gateway")).
+		Returns(http.StatusOK, api.StatusOK, v1alpha1.Gateway{}).
+		Metadata(restfulspec.KeyOpenAPITags, []string{constants.GatewayTag}))
+
+	ws.Route(ws.GET("/namespaces/{namespace}/gateways/{gateway}/pods/{pod}/log").
+		To(handler.PodLog).
+		Doc("Retrieve log of the gateway's pod").
+		Param(ws.PathParameter("namespace", "the watching namespace of the gateway")).
+		Param(ws.PathParameter("pod", "the pod name of the gateway")).
+		Returns(http.StatusOK, api.StatusOK, v1alpha1.Gateway{}).
+		Metadata(restfulspec.KeyOpenAPITags, []string{constants.GatewayTag}))
+
 	container.Add(ws)
 	return nil
 }
--- a/pkg/models/gateway/gateway.go
+++ b/pkg/models/gateway/gateway.go
@@ -19,12 +19,14 @@ package gateway
 import (
 	"context"
 	"fmt"
+	"io"
 	"strings"

 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog"

 	jsonpatch "github.com/evanphx/json-patch"
@@ -38,7 +40,9 @@ import (

 	"kubesphere.io/kubesphere/pkg/api"
 	"kubesphere.io/kubesphere/pkg/apiserver/query"
+	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3"
+	"kubesphere.io/kubesphere/pkg/models/resources/v1alpha3/pod"
 	"kubesphere.io/kubesphere/pkg/simple/client/gateway"
 )

@@ -58,19 +62,25 @@ type GatewayOperator interface {
 	UpdateGateway(namespace string, obj *v1alpha1.Gateway) (*v1alpha1.Gateway, error)
 	UpgradeGateway(namespace string) (*v1alpha1.Gateway, error)
 	ListGateways(query *query.Query) (*api.ListResult, error)
+	GetPods(namesapce string, query *query.Query) (*api.ListResult, error)
+	GetPodLogs(ctx context.Context, namespace string, podName string, logOptions *corev1.PodLogOptions, responseWriter io.Writer) error
 }

 type gatewayOperator struct {
-	client  client.Client
-	cache   cache.Cache
-	options *gateway.Options
+	k8sclient kubernetes.Interface
+	factory   informers.InformerFactory
+	client    client.Client
+	cache     cache.Cache
+	options   *gateway.Options
 }

-func NewGatewayOperator(client client.Client, cache cache.Cache, options *gateway.Options) GatewayOperator {
+func NewGatewayOperator(client client.Client, cache cache.Cache, options *gateway.Options, factory informers.InformerFactory, k8sclient kubernetes.Interface) GatewayOperator {
 	return &gatewayOperator{
-		client:  client,
-		cache:   cache,
-		options: options,
+		client:    client,
+		cache:     cache,
+		options:   options,
+		k8sclient: k8sclient,
+		factory:   factory,
 	}
 }

@@ -449,3 +459,50 @@ func (c *gatewayOperator) filter(object runtime.Object, filter query.Filter) boo
 		return v1alpha3.DefaultObjectMetaFilter(gateway.ObjectMeta, filter)
 	}
 }
+
+func (c *gatewayOperator) GetPods(namesapce string, query *query.Query) (*api.ListResult, error) {
+	podGetter := pod.New(c.factory.KubernetesSharedInformerFactory())
+
+	//TODO: move the selector string to options
+	selector, err := labels.Parse(fmt.Sprintf("app.kubernetes.io/name=ingress-nginx,app.kubernetes.io/instance=kubesphere-router-%s-ingress", namesapce))
+	if err != nil {
+		return nil, fmt.Errorf("invild selector config")
+	}
+	query.LabelSelector = selector.String()
+	return podGetter.List(c.getWorkingNamespace(namesapce), query)
+}
+
+func (c *gatewayOperator) GetPodLogs(ctx context.Context, namespace string, podName string, logOptions *corev1.PodLogOptions, responseWriter io.Writer) error {
+	workingNamespace := c.getWorkingNamespace(namespace)
+
+	pods, err := c.GetPods(namespace, query.New())
+	if err != nil {
+		return err
+	}
+	if !c.hasPod(pods.Items, types.NamespacedName{Namespace: workingNamespace, Name: podName}) {
+		return fmt.Errorf("pod does not exist")
+	}
+
+	podLogRequest := c.k8sclient.CoreV1().
+		Pods(workingNamespace).
+		GetLogs(podName, logOptions)
+	reader, err := podLogRequest.Stream(context.TODO())
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(responseWriter, reader)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *gatewayOperator) hasPod(slice []interface{}, key types.NamespacedName) bool {
+	for _, s := range slice {
+		pod, ok := s.(*corev1.Pod)
+		if ok && client.ObjectKeyFromObject(pod) == key {
+			return true
+		}
+	}
+	return false
+}
--- a/pkg/utils/conversions/core/v1/conversion.go
+++ b/pkg/utils/conversions/core/v1/conversion.go
@@ -0,0 +1,113 @@
+/*
+Copyright The Kubernetes 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.
+*/
+
+// Code generated by conversion-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	url "net/url"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	conversion "k8s.io/apimachinery/pkg/conversion"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+)
+
+// refer to https://github.com/kubernetes/kubernetes/issues/94688
+// conversions must be registered.
+func RegisterConversions(s *runtime.Scheme) error {
+	if err := s.AddGeneratedConversionFunc((*url.Values)(nil), (*v1.PodLogOptions)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_url_Values_To_v1_PodLogOptions(a.(*url.Values), b.(*v1.PodLogOptions), scope)
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func autoConvert_url_Values_To_v1_PodLogOptions(in *url.Values, out *v1.PodLogOptions, s conversion.Scope) error {
+	// WARNING: Field TypeMeta does not have json tag, skipping.
+
+	if values, ok := map[string][]string(*in)["container"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_string(&values, &out.Container, s); err != nil {
+			return err
+		}
+	} else {
+		out.Container = ""
+	}
+	if values, ok := map[string][]string(*in)["follow"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_bool(&values, &out.Follow, s); err != nil {
+			return err
+		}
+	} else {
+		out.Follow = false
+	}
+	if values, ok := map[string][]string(*in)["previous"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_bool(&values, &out.Previous, s); err != nil {
+			return err
+		}
+	} else {
+		out.Previous = false
+	}
+	if values, ok := map[string][]string(*in)["sinceSeconds"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_Pointer_int64(&values, &out.SinceSeconds, s); err != nil {
+			return err
+		}
+	} else {
+		out.SinceSeconds = nil
+	}
+	if values, ok := map[string][]string(*in)["sinceTime"]; ok && len(values) > 0 {
+		if err := metav1.Convert_Slice_string_To_Pointer_v1_Time(&values, &out.SinceTime, s); err != nil {
+			return err
+		}
+	} else {
+		out.SinceTime = nil
+	}
+	if values, ok := map[string][]string(*in)["timestamps"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_bool(&values, &out.Timestamps, s); err != nil {
+			return err
+		}
+	} else {
+		out.Timestamps = false
+	}
+	if values, ok := map[string][]string(*in)["tailLines"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_Pointer_int64(&values, &out.TailLines, s); err != nil {
+			return err
+		}
+	} else {
+		out.TailLines = nil
+	}
+	if values, ok := map[string][]string(*in)["limitBytes"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_Pointer_int64(&values, &out.LimitBytes, s); err != nil {
+			return err
+		}
+	} else {
+		out.LimitBytes = nil
+	}
+	if values, ok := map[string][]string(*in)["insecureSkipTLSVerifyBackend"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_bool(&values, &out.InsecureSkipTLSVerifyBackend, s); err != nil {
+			return err
+		}
+	} else {
+		out.InsecureSkipTLSVerifyBackend = false
+	}
+	return nil
+}
+
+// Convert_url_Values_To_v1_PodLogOptions is an autogenerated conversion function.
+func Convert_url_Values_To_v1_PodLogOptions(in *url.Values, out *v1.PodLogOptions, s conversion.Scope) error {
+	return autoConvert_url_Values_To_v1_PodLogOptions(in, out, s)
+}