diff --git a/pkg/apiserver/apiserver.go b/pkg/apiserver/apiserver.go index 3c72773b9..c05c41c9a 100644 --- 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) { diff --git a/pkg/kapis/gateway/v1alpha1/handler.go b/pkg/kapis/gateway/v1alpha1/handler.go index a31351c12..6e2c0974c 100644 --- 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 + } +} diff --git a/pkg/kapis/gateway/v1alpha1/register.go b/pkg/kapis/gateway/v1alpha1/register.go index c3bc84687..bbcf715af 100644 --- 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 } diff --git a/pkg/models/gateway/gateway.go b/pkg/models/gateway/gateway.go index 4cbbf518a..ec4d4d506 100644 --- 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 +} diff --git a/pkg/utils/conversions/core/v1/conversion.go b/pkg/utils/conversions/core/v1/conversion.go new file mode 100644 index 000000000..3424a7125 --- /dev/null +++ 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) +}