package cli import ( "context" "fmt" models "github.com/portainer/portainer/api/http/models/kubernetes" "github.com/rs/zerolog/log" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" ) // GetServices gets all the services for either at the cluster level or a given namespace in a k8s endpoint. // It returns a list of K8sServiceInfo objects. func (kcl *KubeClient) GetServices(namespace string) ([]models.K8sServiceInfo, error) { if kcl.IsKubeAdmin { return kcl.fetchServices(namespace) } return kcl.fetchServicesForNonAdmin(namespace) } // fetchServicesForNonAdmin gets all the services for either at the cluster level or a given namespace in a k8s endpoint. // the namespace will be coming from NonAdminNamespaces as non-admin users are restricted to certain namespaces. // it returns a list of K8sServiceInfo objects. func (kcl *KubeClient) fetchServicesForNonAdmin(namespace string) ([]models.K8sServiceInfo, error) { log.Debug().Msgf("Fetching services for non-admin user: %v", kcl.NonAdminNamespaces) if len(kcl.NonAdminNamespaces) == 0 { return nil, nil } services, err := kcl.fetchServices(namespace) if err != nil { return nil, err } nonAdminNamespaceSet := kcl.buildNonAdminNamespacesMap() results := make([]models.K8sServiceInfo, 0) for _, service := range services { if _, ok := nonAdminNamespaceSet[service.Namespace]; ok { results = append(results, service) } } return results, nil } // fetchServices gets the services in a given namespace in a k8s endpoint. // It returns a list of K8sServiceInfo objects. func (kcl *KubeClient) fetchServices(namespace string) ([]models.K8sServiceInfo, error) { services, err := kcl.cli.CoreV1().Services(namespace).List(context.TODO(), metav1.ListOptions{}) if err != nil { return nil, err } results := make([]models.K8sServiceInfo, 0) for _, service := range services.Items { results = append(results, parseService(service)) } return results, nil } // parseService converts a k8s native service object to a Portainer K8sServiceInfo object. // service ports, ingress status, labels, annotations, cluster IPs, and external IPs are parsed. // it returns a K8sServiceInfo object. func parseService(service corev1.Service) models.K8sServiceInfo { servicePorts := make([]models.K8sServicePort, 0) for _, port := range service.Spec.Ports { servicePorts = append(servicePorts, models.K8sServicePort{ Name: port.Name, NodePort: int(port.NodePort), Port: int(port.Port), Protocol: string(port.Protocol), TargetPort: port.TargetPort.String(), }) } ingressStatus := make([]models.K8sServiceIngress, 0) for _, status := range service.Status.LoadBalancer.Ingress { ingressStatus = append(ingressStatus, models.K8sServiceIngress{ IP: status.IP, Host: status.Hostname, }) } return models.K8sServiceInfo{ Name: service.Name, UID: string(service.GetUID()), Type: string(service.Spec.Type), Namespace: service.Namespace, CreationDate: service.GetCreationTimestamp().String(), AllocateLoadBalancerNodePorts: service.Spec.AllocateLoadBalancerNodePorts, Ports: servicePorts, IngressStatus: ingressStatus, Labels: service.GetLabels(), Annotations: service.GetAnnotations(), ClusterIPs: service.Spec.ClusterIPs, ExternalName: service.Spec.ExternalName, ExternalIPs: service.Spec.ExternalIPs, Selector: service.Spec.Selector, } } // convertToK8sService converts a K8sServiceInfo object back to a k8s native service object. // this is required for create and update operations. // it returns a v1.Service object. func (kcl *KubeClient) convertToK8sService(info models.K8sServiceInfo) corev1.Service { service := corev1.Service{} service.Name = info.Name service.Spec.Type = corev1.ServiceType(info.Type) service.Namespace = info.Namespace service.Annotations = info.Annotations service.Labels = info.Labels service.Spec.AllocateLoadBalancerNodePorts = info.AllocateLoadBalancerNodePorts service.Spec.Selector = info.Selector for _, p := range info.Ports { port := corev1.ServicePort{} port.Name = p.Name port.NodePort = int32(p.NodePort) port.Port = int32(p.Port) port.Protocol = corev1.Protocol(p.Protocol) port.TargetPort = intstr.FromString(p.TargetPort) service.Spec.Ports = append(service.Spec.Ports, port) } for _, i := range info.IngressStatus { service.Status.LoadBalancer.Ingress = append( service.Status.LoadBalancer.Ingress, corev1.LoadBalancerIngress{IP: i.IP, Hostname: i.Host}, ) } return service } // CreateService creates a new service in a given namespace in a k8s endpoint. func (kcl *KubeClient) CreateService(namespace string, info models.K8sServiceInfo) error { service := kcl.convertToK8sService(info) _, err := kcl.cli.CoreV1().Services(namespace).Create(context.Background(), &service, metav1.CreateOptions{}) return err } // DeleteServices processes a K8sServiceDeleteRequest by deleting each service // in its given namespace. func (kcl *KubeClient) DeleteServices(reqs models.K8sServiceDeleteRequests) error { for namespace := range reqs { for _, service := range reqs[namespace] { err := kcl.cli.CoreV1().Services(namespace).Delete(context.Background(), service, metav1.DeleteOptions{}) if err != nil { return err } } } return nil } // UpdateService updates service in a given namespace in a k8s endpoint. func (kcl *KubeClient) UpdateService(namespace string, info models.K8sServiceInfo) error { service := kcl.convertToK8sService(info) _, err := kcl.cli.CoreV1().Services(namespace).Update(context.Background(), &service, metav1.UpdateOptions{}) return err } // CombineServicesWithApplications retrieves applications based on service selectors in a given namespace // for all services, it lists pods based on the service selector and converts the pod to an application // if replicasets are found, it updates the owner reference to deployment // it then combines the service with the application // finally, it returns a list of K8sServiceInfo objects func (kcl *KubeClient) CombineServicesWithApplications(services []models.K8sServiceInfo) ([]models.K8sServiceInfo, error) { if containsServiceWithSelector(services) { updatedServices := make([]models.K8sServiceInfo, len(services)) pods, replicaSets, _, _, _, _, err := kcl.fetchAllPodsAndReplicaSets("", metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("an error occurred during the CombineServicesWithApplications operation, unable to fetch pods and replica sets. Error: %w", err) } for index, service := range services { updatedService := service application, err := kcl.GetApplicationFromServiceSelector(pods, service, replicaSets) if err != nil { return services, fmt.Errorf("an error occurred during the CombineServicesWithApplications operation, unable to get application from service. Error: %w", err) } if application != nil { updatedService.Applications = append(updatedService.Applications, *application) } updatedServices[index] = updatedService } return updatedServices, nil } return services, nil } // containsServiceWithSelector checks if a list of services contains a service with a selector // it returns true if any service has a selector, otherwise false func containsServiceWithSelector(services []models.K8sServiceInfo) bool { for _, service := range services { if len(service.Selector) > 0 { return true } } return false } // buildServicesMap builds a map of service names from a list of K8sServiceInfo objects // it returns a map of service names for lookups func (kcl *KubeClient) buildServicesMap(services []models.K8sServiceInfo) map[string]struct{} { serviceMap := make(map[string]struct{}) for _, service := range services { serviceMap[service.Name] = struct{}{} } return serviceMap }