Merge pull request #234 from devtron-labs/fix-kubelink-cache-issue-main

fix: high memory usage in kubelink informers

Author: Ash-exp

kubelink/internals/middleware/instrument.go

@@ -27,9 +27,10 @@ import (
 
 // metrics name constant
 const (
-	KUBELINK_HTTP_DURATION_SECONDS = "kubelink_http_duration_seconds"
-	KUBELINK_HTTP_REQUESTS_TOTAL   = "kubelink_http_requests_total"
-	KUBELINK_HTTP_REQUESTS_CURRENT = "kubelink_http_requests_current"
+	KUBELINK_HTTP_DURATION_SECONDS                    = "kubelink_http_duration_seconds"
+	KUBELINK_HTTP_REQUESTS_TOTAL                      = "kubelink_http_requests_total"
+	KUBELINK_HTTP_REQUESTS_CURRENT                    = "kubelink_http_requests_current"
+	KUBELINK_INFORMER_DATA_TRANSFORM_DURATION_SECONDS = "kubelink_informer_data_transform_duration_seconds"
 )
 
 // metrics labels constants
@@ -57,6 +58,17 @@ var currentRequestGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Help: "no of request being served currently",
 }, []string{PATH, METHOD})
 
+var InformerDataTransformDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Name: KUBELINK_INFORMER_DATA_TRANSFORM_DURATION_SECONDS,
+	Help: "Duration of informer data transform request",
+}, []string{CLUSTER_NAME, NAMESPACE, RELEASE_NAME})
+
+const (
+	CLUSTER_NAME  = "clusterName"
+	NAMESPACE     = "namespace"
+	RELEASE_NAME  = "releaseName"
+)
+
 // prometheusMiddleware implements mux.MiddlewareFunc.
 func PrometheusMiddleware(next http.Handler) http.Handler {
 	//	prometheus.MustRegister(requestCounter)

kubelink/pkg/k8sInformer/K8sInformer.go

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 k8sInformer
+
+import (
+	"fmt"
+	client "github.com/devtron-labs/kubelink/grpc"
+)
+
+const (
+	secretKeyAppDetailKey = "appDetail"
+)
+
+type DeployedAppDetailDto struct {
+	*client.DeployedAppDetail
+}
+
+func NewDeployedAppDetailDto(appDetail *client.DeployedAppDetail) *DeployedAppDetailDto {
+	return &DeployedAppDetailDto{DeployedAppDetail: appDetail}
+}
+
+func (r *DeployedAppDetailDto) getUniqueReleaseIdentifier() string {
+	if r == nil || r.EnvironmentDetail == nil {
+		return ""
+	}
+	// adding cluster id with release name and namespace because there can be case when two cluster or two namespaces have release with same name
+	return fmt.Sprintf("%s_%s_%d", r.EnvironmentDetail.Namespace, r.AppName, r.EnvironmentDetail.ClusterId)
+}

kubelink/pkg/k8sInformer/bean.go

@@ -18,17 +18,11 @@ package k8sInformer
 
 import (
 	"errors"
-	"helm.sh/helm/v3/pkg/release"
 )
 
-type ReleaseDto struct {
-	*release.Release
-}
-
-func (r *ReleaseDto) getUniqueReleaseIdentifier() string {
-	return r.Namespace + "_" + r.Name
-}
-
 var (
+	// ErrorCacheMissReleaseNotFound is returned when a release is not found in the cache
 	ErrorCacheMissReleaseNotFound = errors.New("release not found in cache")
+	// InformerAlreadyExistError is returned when an informer already exists
+	InformerAlreadyExistError = errors.New(INFORMER_ALREADY_EXIST_MESSAGE)
 )

kubelink/pkg/k8sInformer/constants.go kubelink/pkg/k8sInformer/error.gokubelink/pkg/k8sInformer/constants.go renamed to kubelink/pkg/k8sInformer/error.go

@@ -0,0 +1,84 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 k8sInformer
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	client "github.com/devtron-labs/kubelink/grpc"
+	"github.com/devtron-labs/kubelink/pkg/service/helmApplicationService/release"
+	"io"
+)
+
+func parseSecretDataForDeployedAppDetail(appDetail *client.DeployedAppDetail) (map[string][]byte, error) {
+	appDetailBytes, err := json.Marshal(appDetail)
+	if err != nil {
+		return nil, err
+	}
+	data := make(map[string][]byte)
+	data[secretKeyAppDetailKey] = appDetailBytes
+	return data, nil
+}
+
+func getDeployedAppDetailFromSecretData(data map[string][]byte) (*client.DeployedAppDetail, error) {
+	if appDetailBytes, ok := data[secretKeyAppDetailKey]; ok {
+		var appDetail client.DeployedAppDetail
+		err := json.Unmarshal(appDetailBytes, &appDetail)
+		if err != nil {
+			return nil, err
+		}
+		return &appDetail, nil
+	}
+	return nil, errors.New("app detail not found in secret data")
+}
+
+func decodeHelmReleaseData(data string) (*release.Release, error) {
+	// base64 decode string
+	b64 := base64.StdEncoding
+	b, err := b64.DecodeString(data)
+	if err != nil {
+		return nil, err
+	}
+
+	var magicGzip = []byte{0x1f, 0x8b, 0x08}
+
+	// For backwards compatibility with releases that were stored before
+	// compression was introduced we skip decompression if the
+	// gzip magic header is not found
+	if len(b) > 3 && bytes.Equal(b[0:3], magicGzip) {
+		r, err := gzip.NewReader(bytes.NewReader(b))
+		if err != nil {
+			return nil, err
+		}
+		defer r.Close()
+		b2, err := io.ReadAll(r)
+		if err != nil {
+			return nil, err
+		}
+		b = b2
+	}
+
+	var rls release.Release
+	// unmarshal release object bytes
+	if err := json.Unmarshal(b, &rls); err != nil {
+		return nil, err
+	}
+	return &rls, nil
+}

kubelink/pkg/k8sInformer/util.go

@@ -2,11 +2,13 @@ package adapter
 
 import (
 	"github.com/devtron-labs/common-lib/helmLib/registry"
-	"github.com/devtron-labs/common-lib/utils/remoteConnection/bean"
+	remoteConnection "github.com/devtron-labs/common-lib/utils/remoteConnection/bean"
 	client "github.com/devtron-labs/kubelink/grpc"
+	"github.com/devtron-labs/kubelink/pkg/service/helmApplicationService/release"
 	"github.com/devtron-labs/kubelink/pkg/util"
 	"google.golang.org/protobuf/types/known/timestamppb"
-	"helm.sh/helm/v3/pkg/release"
+	helmChart "helm.sh/helm/v3/pkg/chart"
+	helmRelease "helm.sh/helm/v3/pkg/release"
 )
 
 func NewRegistryConfig(credential *client.RegistryCredential) (*registry.Configuration, error) {
@@ -36,36 +38,36 @@ func NewRegistryConfig(credential *client.RegistryCredential) (*registry.Configu
 
 		connectionConfig := credential.RemoteConnectionConfig
 		if connectionConfig != nil {
-			registryConfig.RemoteConnectionConfig = &bean.RemoteConnectionConfigBean{}
+			registryConfig.RemoteConnectionConfig = &remoteConnection.RemoteConnectionConfigBean{}
 			switch connectionConfig.RemoteConnectionMethod {
 			case client.RemoteConnectionMethod_DIRECT:
-				registryConfig.RemoteConnectionConfig.ConnectionMethod = bean.RemoteConnectionMethodDirect
+				registryConfig.RemoteConnectionConfig.ConnectionMethod = remoteConnection.RemoteConnectionMethodDirect
 			case client.RemoteConnectionMethod_PROXY:
-				registryConfig.RemoteConnectionConfig.ConnectionMethod = bean.RemoteConnectionMethodProxy
+				registryConfig.RemoteConnectionConfig.ConnectionMethod = remoteConnection.RemoteConnectionMethodProxy
 				registryConfig.RemoteConnectionConfig.ProxyConfig = ConvertConfigToProxyConfig(connectionConfig)
 			case client.RemoteConnectionMethod_SSH:
-				registryConfig.RemoteConnectionConfig.ConnectionMethod = bean.RemoteConnectionMethodSSH
+				registryConfig.RemoteConnectionConfig.ConnectionMethod = remoteConnection.RemoteConnectionMethodSSH
 				registryConfig.RemoteConnectionConfig.SSHTunnelConfig = ConvertConfigToSSHTunnelConfig(connectionConfig)
 			}
 		}
 	}
 	return registryConfig, nil
 }
 
-func ConvertConfigToProxyConfig(config *client.RemoteConnectionConfig) *bean.ProxyConfig {
-	var proxyConfig *bean.ProxyConfig
+func ConvertConfigToProxyConfig(config *client.RemoteConnectionConfig) *remoteConnection.ProxyConfig {
+	var proxyConfig *remoteConnection.ProxyConfig
 	if config.ProxyConfig != nil {
-		proxyConfig = &bean.ProxyConfig{
+		proxyConfig = &remoteConnection.ProxyConfig{
 			ProxyUrl: config.ProxyConfig.ProxyUrl,
 		}
 	}
 	return proxyConfig
 }
 
-func ConvertConfigToSSHTunnelConfig(config *client.RemoteConnectionConfig) *bean.SSHTunnelConfig {
-	var sshConfig *bean.SSHTunnelConfig
+func ConvertConfigToSSHTunnelConfig(config *client.RemoteConnectionConfig) *remoteConnection.SSHTunnelConfig {
+	var sshConfig *remoteConnection.SSHTunnelConfig
 	if config.SSHTunnelConfig != nil {
-		sshConfig = &bean.SSHTunnelConfig{
+		sshConfig = &remoteConnection.SSHTunnelConfig{
 			SSHUsername:      config.SSHTunnelConfig.SSHUsername,
 			SSHPassword:      config.SSHTunnelConfig.SSHPassword,
 			SSHAuthKey:       config.SSHTunnelConfig.SSHAuthKey,
@@ -119,3 +121,46 @@ func GetAppDetailRequestFromGetResourceTreeRequest(req *client.GetResourceTreeRe
 		CacheConfig:   req.CacheConfig,
 	}
 }
+
+func NewRelease(helmRelease *helmRelease.Release) *release.Release {
+	if helmRelease == nil {
+		return nil
+	}
+	return &release.Release{
+		Name:      helmRelease.Name,
+		Namespace: helmRelease.Namespace,
+		Info:      NewReleaseInfo(helmRelease.Info),
+		Chart:     NewChart(helmRelease.Chart),
+	}
+}
+
+func NewReleaseInfo(helmReleaseInfo *helmRelease.Info) *release.Info {
+	if helmReleaseInfo == nil {
+		return nil
+	}
+	return &release.Info{
+		LastDeployed: helmReleaseInfo.LastDeployed,
+		Status:       helmReleaseInfo.Status,
+	}
+}
+
+func NewChart(helmChart *helmChart.Chart) *release.Chart {
+	if helmChart == nil {
+		return nil
+	}
+	return &release.Chart{
+		Metadata: NewChartMetadata(helmChart.Metadata),
+	}
+}
+
+func NewChartMetadata(helmChartMetadata *helmChart.Metadata) *release.Metadata {
+	if helmChartMetadata == nil {
+		return nil
+	}
+	return &release.Metadata{
+		Name:    helmChartMetadata.Name,
+		Version: helmChartMetadata.Version,
+		Icon:    helmChartMetadata.Icon,
+		Home:    helmChartMetadata.Home,
+	}
+}