Enhancement: Kubernetes api refactoring

Wire.go

@@ -58,6 +58,7 @@ import (
 	"github.com/devtron-labs/devtron/client/gitSensor"
 	"github.com/devtron-labs/devtron/client/grafana"
 	jClient "github.com/devtron-labs/devtron/client/jira"
+	util4 "github.com/devtron-labs/devtron/client/k8s/application/util"
 	"github.com/devtron-labs/devtron/client/lens"
 	"github.com/devtron-labs/devtron/client/telemetry"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
@@ -126,6 +127,7 @@ func InitializeApp() (*App, error) {
 		externalLink.ExternalLinkWireSet,
 		team.TeamsWireSet,
 		AuthWireSet,
+		util4.NewK8sUtil,
 		user.UserWireSet,
 		sso.SsoConfigWireSet,
 		cluster.ClusterWireSet,
@@ -358,7 +360,6 @@ func InitializeApp() (*App, error) {
 		wire.Bind(new(repository8.ImageTaggingRepository), new(*repository8.ImageTaggingRepositoryImpl)),
 		pipeline.NewImageTaggingServiceImpl,
 		wire.Bind(new(pipeline.ImageTaggingService), new(*pipeline.ImageTaggingServiceImpl)),
-		util.NewK8sUtil,
 		argocdServer.NewVersionServiceImpl,
 		wire.Bind(new(argocdServer.VersionService), new(*argocdServer.VersionServiceImpl)),
 

api/cluster/EnvironmentRestHandler.go

@@ -20,9 +20,7 @@ package cluster
 import (
 	"context"
 	"encoding/json"
-	"github.com/devtron-labs/devtron/internal/util"
-	"github.com/devtron-labs/devtron/util/k8s"
-	"k8s.io/client-go/kubernetes"
+	"github.com/devtron-labs/devtron/client/k8s/application"
 	"net/http"
 	"strconv"
 	"strings"
@@ -60,7 +58,7 @@ type EnvironmentRestHandler interface {
 
 type EnvironmentRestHandlerImpl struct {
 	environmentClusterMappingsService request.EnvironmentService
-	k8sApplicationService             k8s.K8sApplicationService
+	k8sClientService                  application.K8sClientService
 	logger                            *zap.SugaredLogger
 	userService                       user.UserService
 	validator                         *validator.Validate
@@ -74,10 +72,7 @@ type ClusterReachableResponse struct {
 	ClusterName      string `json:"clusterName"`
 }
 
-func NewEnvironmentRestHandlerImpl(svc request.EnvironmentService, k8sApplicationService k8s.K8sApplicationService, logger *zap.SugaredLogger, userService user.UserService,
-	validator *validator.Validate, enforcer casbin.Enforcer,
-	deleteService delete2.DeleteService,
-) *EnvironmentRestHandlerImpl {
+func NewEnvironmentRestHandlerImpl(svc request.EnvironmentService, logger *zap.SugaredLogger, userService user.UserService, validator *validator.Validate, enforcer casbin.Enforcer, deleteService delete2.DeleteService, k8sClientService application.K8sClientService) *EnvironmentRestHandlerImpl {
 	cfg := &bean.Config{}
 	err := env.Parse(cfg)
 	if err != nil {
@@ -87,7 +82,7 @@ func NewEnvironmentRestHandlerImpl(svc request.EnvironmentService, k8sApplicatio
 	logger.Infow("evironment rest handler initialized", "ignoreAuthCheckValue", cfg.IgnoreAuthCheck)
 	return &EnvironmentRestHandlerImpl{
 		environmentClusterMappingsService: svc,
-		k8sApplicationService:             k8sApplicationService,
+		k8sClientService:                  k8sClientService,
 		logger:                            logger,
 		userService:                       userService,
 		validator:                         validator,
@@ -514,29 +509,23 @@ func (impl EnvironmentRestHandlerImpl) GetEnvironmentConnection(w http.ResponseW
 	}
 	//RBAC enforcer Ends
 	// getting restConfig and clientSet outside the goroutine because we don't want to call goroutine func with receiver function
-	restConfig, err := impl.k8sApplicationService.GetRestConfigByClusterId(context.Background(), clusterBean.Id)
+	restConfig, err := impl.k8sClientService.GetRestConfigByClusterId(context.Background(), clusterBean.Id)
 	if err != nil {
 		impl.logger.Errorw("error in getting restConfig by cluster", "err", err, "clusterId", clusterBean.Id)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
 	}
-	k8sHttpClient, err := util.OverrideK8sHttpClientWithTracer(restConfig)
+	k8sClientSet, err := impl.k8sClientService.CreateK8sClientSet(restConfig)
 	if err != nil {
-		impl.logger.Errorw("service err, OverrideK8sHttpClientWithTracer", "err", err, "restConfig", restConfig)
-		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
-	}
-	k8sClientSet, err := kubernetes.NewForConfigAndClient(restConfig, k8sHttpClient)
-	if err != nil {
-		impl.logger.Errorw("error in getting client set by rest config", "err", err, "restConfig", restConfig)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
 	}
+
 	responseObj := &ClusterReachableResponse{
 		ClusterReachable: true,
 		ClusterName:      clusterBean.ClusterName,
 	}
-	err = impl.k8sApplicationService.FetchConnectionStatusForCluster(k8sClientSet, clusterBean.Id)
+	err = impl.k8sClientService.FetchConnectionStatusForCluster(k8sClientSet, clusterBean.Id)
 	if err != nil {
 		responseObj.ClusterReachable = false
 	}

api/helm-app/HelmAppService.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	util3 "github.com/devtron-labs/devtron/client/k8s/application/util"
 	"net/http"
 	"reflect"
 	"strconv"
@@ -78,7 +79,7 @@ type HelmAppServiceImpl struct {
 	installedAppRepository               repository.InstalledAppRepository
 	appRepository                        app.AppRepository
 	clusterRepository                    clusterRepository.ClusterRepository
-	K8sUtil                              *util.K8sUtil
+	K8sUtil                              *util3.K8sUtil
 	helmReleaseConfig                    *HelmReleaseConfig
 }
 
@@ -88,7 +89,7 @@ func NewHelmAppServiceImpl(Logger *zap.SugaredLogger, clusterService cluster.Clu
 	appStoreApplicationVersionRepository appStoreDiscoverRepository.AppStoreApplicationVersionRepository,
 	environmentService cluster.EnvironmentService, pipelineRepository pipelineConfig.PipelineRepository,
 	installedAppRepository repository.InstalledAppRepository, appRepository app.AppRepository,
-	clusterRepository clusterRepository.ClusterRepository, K8sUtil *util.K8sUtil,
+	clusterRepository clusterRepository.ClusterRepository, K8sUtil *util3.K8sUtil,
 	helmReleaseConfig *HelmReleaseConfig) *HelmAppServiceImpl {
 	return &HelmAppServiceImpl{
 		logger:                               Logger,
@@ -141,7 +142,7 @@ func (impl *HelmAppServiceImpl) listApplications(ctx context.Context, clusterIds
 	for _, clusterDetail := range clusters {
 		config := &ClusterConfig{
 			ApiServerUrl: clusterDetail.ServerUrl,
-			Token:        clusterDetail.Config[util.BearerToken],
+			Token:        clusterDetail.Config[util3.BearerToken],
 			ClusterId:    int32(clusterDetail.Id),
 			ClusterName:  clusterDetail.ClusterName,
 		}
@@ -266,7 +267,7 @@ func (impl *HelmAppServiceImpl) GetClusterConf(clusterId int) (*ClusterConfig, e
 	}
 	config := &ClusterConfig{
 		ApiServerUrl: cluster.ServerUrl,
-		Token:        cluster.Config[util.BearerToken],
+		Token:        cluster.Config[util3.BearerToken],
 		ClusterId:    int32(cluster.Id),
 		ClusterName:  cluster.ClusterName,
 	}

client/k8s/application/Application.go

@@ -3,8 +3,10 @@ package application
 import (
 	"context"
 	"encoding/json"
-	"github.com/devtron-labs/devtron/internal/util"
-	"github.com/devtron-labs/devtron/pkg/cluster/repository"
+	"fmt"
+	"github.com/devtron-labs/devtron/client/k8s/application/util"
+	"github.com/devtron-labs/devtron/pkg/cluster"
+	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
 	"io"
 	apiv1 "k8s.io/api/core/v1"
@@ -15,10 +17,13 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/utils/pointer"
+	"log"
 	"net/http"
+	"net/url"
 	"strings"
 )
 
@@ -33,17 +38,22 @@ type K8sClientService interface {
 	GetResourceList(ctx context.Context, restConfig *rest.Config, request *K8sRequestBean) (*ResourceListResponse, bool, error)
 	ApplyResource(ctx context.Context, restConfig *rest.Config, request *K8sRequestBean, manifest string) (*ManifestResponse, error)
 	PatchResource(ctx context.Context, restConfig *rest.Config, pt types.PatchType, request *K8sRequestBean, manifest string) (*ManifestResponse, error)
+	GetRestConfigByClusterId(ctx context.Context, clusterId int) (*rest.Config, error)
+	FetchConnectionStatusForCluster(k8sClientSet *kubernetes.Clientset, clusterId int) error
+	CreateK8sClientSet(restConfig *rest.Config) (*kubernetes.Clientset, error)
 }
 
 type K8sClientServiceImpl struct {
-	logger            *zap.SugaredLogger
-	clusterRepository repository.ClusterRepository
+	logger         *zap.SugaredLogger
+	clusterService cluster.ClusterService
+	k8sUtil        *util.K8sUtil
 }
 
-func NewK8sClientServiceImpl(logger *zap.SugaredLogger, clusterRepository repository.ClusterRepository) *K8sClientServiceImpl {
+func NewK8sClientServiceImpl(logger *zap.SugaredLogger, clusterService cluster.ClusterService, k8sUtil *util.K8sUtil) *K8sClientServiceImpl {
 	return &K8sClientServiceImpl{
-		logger:            logger,
-		clusterRepository: clusterRepository,
+		logger:         logger,
+		clusterService: clusterService,
+		k8sUtil:        k8sUtil,
 	}
 }
 
@@ -80,6 +90,19 @@ type ResourceListResponse struct {
 	Resources unstructured.UnstructuredList `json:"resources,omitempty"`
 }
 
+func (impl *K8sClientServiceImpl) GetRestConfigByClusterId(ctx context.Context, clusterId int) (*rest.Config, error) {
+	_, span := otel.Tracer("orchestrator").Start(ctx, "K8sApplicationService.GetRestConfigByClusterId")
+	defer span.End()
+	cluster, err := impl.clusterService.FindById(clusterId)
+	if err != nil {
+		impl.logger.Errorw("error in getting cluster by ID", "err", err, "clusterId")
+		return nil, err
+	}
+	clusterConfig := cluster.GetClusterConfig()
+	restConfig, err := impl.k8sUtil.GetRestConfigByCluster(&clusterConfig)
+	return restConfig, nil
+}
+
 func (impl K8sClientServiceImpl) GetResource(ctx context.Context, restConfig *rest.Config, request *K8sRequestBean) (*ManifestResponse, error) {
 	resourceIf, namespaced, err := impl.GetResourceIf(restConfig, request)
 	if err != nil {
@@ -447,3 +470,46 @@ func (impl K8sClientServiceImpl) PatchResource(ctx context.Context, restConfig *
 func (impl K8sClientServiceImpl) ApplyResource(ctx context.Context, restConfig *rest.Config, request *K8sRequestBean, manifest string) (*ManifestResponse, error) {
 	return impl.PatchResource(ctx, restConfig, types.StrategicMergePatchType, request, manifest)
 }
+
+func (impl *K8sClientServiceImpl) FetchConnectionStatusForCluster(k8sClientSet *kubernetes.Clientset, clusterId int) error {
+	//using livez path as healthz path is deprecated
+	path := util.LiveZ
+	response, err := k8sClientSet.Discovery().RESTClient().Get().AbsPath(path).DoRaw(context.Background())
+	log.Println("received response for cluster livez status", "response", string(response), "err", err, "clusterId", clusterId)
+	if err != nil {
+		if _, ok := err.(*url.Error); ok {
+			err = fmt.Errorf("Incorrect server url : %v", err)
+		} else if statusError, ok := err.(*errors.StatusError); ok {
+			if statusError != nil {
+				errReason := statusError.ErrStatus.Reason
+				var errMsg string
+				if errReason == metav1.StatusReasonUnauthorized {
+					errMsg = "token seems invalid or does not have sufficient permissions"
+				} else {
+					errMsg = statusError.ErrStatus.Message
+				}
+				err = fmt.Errorf("%s : %s", errReason, errMsg)
+			} else {
+				err = fmt.Errorf("Validation failed : %v", err)
+			}
+		} else {
+			err = fmt.Errorf("Validation failed : %v", err)
+		}
+	} else if err == nil && string(response) != "ok" {
+		err = fmt.Errorf("Validation failed with response : %s", string(response))
+	}
+	return err
+}
+func (impl *K8sClientServiceImpl) CreateK8sClientSet(restConfig *rest.Config) (*kubernetes.Clientset, error) {
+	k8sHttpClient, err := util.OverrideK8sHttpClientWithTracer(restConfig)
+	if err != nil {
+		impl.logger.Errorw("service err, OverrideK8sHttpClientWithTracer", "err", err, "restConfig", restConfig)
+		return nil, err
+	}
+	k8sClientSet, err := kubernetes.NewForConfigAndClient(restConfig, k8sHttpClient)
+	if err != nil {
+		impl.logger.Errorw("error in getting client set by rest config", "err", err, "restConfig", restConfig)
+		return nil, err
+	}
+	return k8sClientSet, err
+}