Add DB support for controlplane data in dataplane side

adapter/internal/operator/controllers/dp/api_controller.go

@@ -2017,7 +2017,7 @@ func (apiReconciler *APIReconciler) handleOwnerReference(ctx context.Context, ob
 		if err := apiReconciler.client.Get(ctx, req.NamespacedName, &apiCR); err == nil {
 			apis = append(apis, apiCR)
 		} else {
-			loggers.LoggerAPKOperator.Errorf("Error while loading api: %+v", req)
+			loggers.LoggerAPKOperator.Errorf("Error while loading api: %+v, Error: %v", req, err)
 		}
 	}
 	// Prepare owner references for the route

common-controller/cmd/main.go

@@ -21,6 +21,7 @@ import (
 	logger "github.com/sirupsen/logrus"
 	commoncontroller "github.com/wso2/apk/common-controller/commoncontroller"
 	config "github.com/wso2/apk/common-controller/internal/config"
+	"github.com/wso2/apk/common-controller/internal/database"
 	"github.com/wso2/apk/common-controller/internal/server"
 	web "github.com/wso2/apk/common-controller/internal/web"
 )
@@ -30,6 +31,11 @@ func main() {
 	logger.Info("Starting the Web server")
 	go web.StartWebServer()
 	go server.StartInternalServer()
+	if conf.CommonController.Database.Enabled {
+		logger.Info("Starting the Database connection")
+		database.ConnectToDB()
+	}
+	defer database.CloseDBConn()
 	logger.Info("Starting the Common Controller")
 	commoncontroller.InitCommonControllerServer(conf)
 }

common-controller/go.mod

@@ -3,6 +3,7 @@ module github.com/wso2/apk/common-controller
 go 1.20
 
 require (
+	github.com/jackc/pgx/v5 v5.5.2
 	github.com/onsi/ginkgo/v2 v2.11.0
 	github.com/onsi/gomega v1.27.10
 	github.com/prometheus/client_golang v1.18.0
@@ -20,7 +21,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/pelletier/go-toml v1.9.5
 	github.com/redis/go-redis/v9 v9.2.1
-	github.com/wso2/apk/adapter v0.0.0-20231207051518-6dd728943082
+	github.com/wso2/apk/adapter v0.0.0-20231214082511-af2c8b8a19f1
 	github.com/wso2/apk/common-go-libs v0.0.0-20240304050809-a382bc6b0d82
 	google.golang.org/grpc v1.62.0
 )
@@ -42,6 +43,9 @@ require (
 	github.com/go-playground/validator/v10 v10.14.0 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
 	github.com/leodido/go-urn v1.2.4 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
@@ -61,6 +65,7 @@ require (
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/crypto v0.19.0 // indirect
 	golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb // indirect
+	golang.org/x/sync v0.6.0 // indirect
 )
 
 require (

common-controller/go.sum

@@ -121,6 +121,14 @@ github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
 github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.5.2 h1:iLlpgp4Cp/gC9Xuscl7lFL1PhhW+ZLtXZcrfCt4C3tA=
+github.com/jackc/pgx/v5 v5.5.2/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
@@ -285,6 +293,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

common-controller/internal/config/default_config.go

@@ -48,5 +48,21 @@ var defaultConfig = &Config{
 			Type:    "prometheus",
 			Port:    18006,
 		},
+		Database: database{
+			Enabled:  false,
+			Name:     "DATAPLANE",
+			Username: "wso2carbon",
+			Password: "wso2carbon",
+			Host:     "wso2apk-db-service.apk",
+			Port:     5432,
+			PoolOptions: dbPool{
+				PoolMaxConns:              4,
+				PoolMinConns:              0,
+				PoolMaxConnLifetime:       "1h",
+				PoolMaxConnIdleTime:       "1h",
+				PoolHealthCheckPeriod:     "1m",
+				PoolMaxConnLifetimeJitter: "1s",
+			},
+		},
 	},
 }

common-controller/internal/config/types.go

@@ -49,6 +49,7 @@ type commoncontroller struct {
 	InternalAPIServer internalAPIServer
 	ControlPlane      controlplane
 	Metrics           metrics
+	Database          database
 }
 type controlplane struct {
 	Enabled             bool
@@ -110,3 +111,36 @@ type metrics struct {
 	Port               int32
 	CollectionInterval int32
 }
+
+type database struct {
+	Enabled     bool
+	Name        string
+	Username    string
+	Password    string
+	Host        string
+	Port        int
+	PoolOptions dbPool
+}
+
+type dbPool struct {
+	// PoolMaxConns is the maximum size of the pool. The default is the greater of 4 or runtime.NumCPU()
+	PoolMaxConns int
+
+	// PoolMinConns is the minimum size of the pool. After connection closes, the pool might dip below MinConns. A low
+	// number of MinConns might mean the pool is empty after MaxConnLifetime until the health check has a chance
+	// to create new connections.
+	PoolMinConns int
+
+	// PoolMaxConnLifetime is the duration since creation after which a connection will be automatically closed.
+	PoolMaxConnLifetime string
+
+	// PoolMaxConnIdleTime is the duration after which an idle connection will be automatically closed by the health check.
+	PoolMaxConnIdleTime string
+
+	// HealthCheckPeriod is the duration between checks of the health of idle connections.
+	PoolHealthCheckPeriod string
+
+	// PoolMaxConnLifetimeJitter is the duration after MaxConnLifetime to randomly decide to close a connection.
+	// This helps prevent all connections from being closed at the exact same time, starving the pool.
+	PoolMaxConnLifetimeJitter string
+}

common-controller/internal/controlplane/artifact_deployer.go

@@ -27,6 +27,7 @@ type ArtifactDeployer interface {
 	DeployApplicationMappings(applicationMapping server.ApplicationMapping) error
 	UpdateApplicationMappings(applicationMapping server.ApplicationMapping) error
 	DeployKeyMappings(keyMapping server.ApplicationKeyMapping) error
+	UpdateKeyMappings(keyMapping server.ApplicationKeyMapping) error
 	GetApplication(applicationID string) (server.Application, error)
 	GetSubscription(subscriptionID string) (server.Subscription, error)
 	GetApplicationMappings(applicationID string) (server.ApplicationMapping, error)

common-controller/internal/controlplane/controlplane_client.go

@@ -167,7 +167,7 @@ func (controlPlaneGrpcClient *Agent) initializeGrpcStreaming() *grpc.ClientConn
 }
 func (controlPlaneGrpcClient *Agent) handleEvents(event *subscription.Event) {
 	loggers.LoggerAPKOperator.Infof("Received event %s", event.Type)
-	if event.Type == constants.AllEvnts {
+	if event.Type == constants.AllEvents {
 		go controlPlaneGrpcClient.retrieveAllData()
 	} else if event.Type == constants.ApplicationCreated {
 		loggers.LoggerAPKOperator.Infof("Received APPLICATION_CREATED event.")
@@ -296,7 +296,19 @@ func (controlPlaneGrpcClient *Agent) handleEvents(event *subscription.Event) {
 			loggers.LoggerAPKOperator.Infof("Received ApplicationMapping %s", applicationMapping.UUID)
 			controlPlaneGrpcClient.artifactDeployer.UpdateApplicationMappings(applicationMapping)
 		}
-
+	} else if event.Type == constants.ApplicationKeyMappingUpdated {
+		loggers.LoggerAPKOperator.Infof("Received APPLICATION_KEY_MAPPING_UPDATED event.")
+		if event.ApplicationKeyMapping != nil {
+			applicationKeyMapping := server.ApplicationKeyMapping{ApplicationUUID: event.ApplicationKeyMapping.ApplicationUUID,
+				SecurityScheme:        event.ApplicationKeyMapping.SecurityScheme,
+				ApplicationIdentifier: event.ApplicationKeyMapping.ApplicationIdentifier,
+				KeyType:               event.ApplicationKeyMapping.KeyType,
+				EnvID:                 event.ApplicationKeyMapping.EnvID,
+				OrganizationID:        event.ApplicationKeyMapping.Organization,
+			}
+			loggers.LoggerAPKOperator.Infof("Received ApplicationKeyMapping %s", applicationKeyMapping.ApplicationUUID)
+			controlPlaneGrpcClient.artifactDeployer.UpdateKeyMappings(applicationKeyMapping)
+		}
 	}
 }
 func (controlPlaneGrpcClient *Agent) retrieveAllData() {
@@ -429,20 +441,28 @@ func (controlPlaneGrpcClient *Agent) retrieveDataFromResponseChannel(response re
 			loggers.LoggerAPI.Infof("Received Subscription information.")
 			subList := newResponse.(*SubscriptionList)
 			resolvedSubscriptionList := marshalMultipleSubscriptions(subList)
-			controlPlaneGrpcClient.artifactDeployer.DeployAllSubscriptions(resolvedSubscriptionList)
+			if len(resolvedSubscriptionList.List) > 0 {
+				controlPlaneGrpcClient.artifactDeployer.DeployAllSubscriptions(resolvedSubscriptionList)
+			}
 
 		case *ApplicationList:
 			loggers.LoggerAPI.Infof("Received Application information.")
 			appList := newResponse.(*ApplicationList)
 			resolvedApplicationList := marshalMultipleApplications(appList)
 			resolvedApplicationKeyMappingList := marshalMultipleApplicationKeyMappings(appList)
-			controlPlaneGrpcClient.artifactDeployer.DeployAllApplications(resolvedApplicationList)
-			controlPlaneGrpcClient.artifactDeployer.DeployAllKeyMappings(resolvedApplicationKeyMappingList)
+			if len(resolvedApplicationList.List) > 0 {
+				controlPlaneGrpcClient.artifactDeployer.DeployAllApplications(resolvedApplicationList)
+			}
+			if len(resolvedApplicationKeyMappingList.List) > 0 {
+				controlPlaneGrpcClient.artifactDeployer.DeployAllKeyMappings(resolvedApplicationKeyMappingList)
+			}
 		case *ApplicationMappingList:
 			loggers.LoggerAPI.Infof("Received Application Mapping information.")
 			appMappingList := newResponse.(*ApplicationMappingList)
 			resolvedApplicationMappingList := marshalMultipleApplicationMappings(appMappingList)
-			controlPlaneGrpcClient.artifactDeployer.DeployAllApplicationMappings(resolvedApplicationMappingList)
+			if len(resolvedApplicationMappingList.List) > 0 {
+				controlPlaneGrpcClient.artifactDeployer.DeployAllApplicationMappings(resolvedApplicationMappingList)
+			}
 		default:
 			loggers.LoggerAPI.Debugf("Unknown type %T", t)
 		}

common-controller/internal/controlplane/k8s_artifact_deployer.go

@@ -24,6 +24,7 @@ import (
 	"github.com/wso2/apk/common-controller/internal/loggers"
 	"github.com/wso2/apk/common-controller/internal/server"
 	cpv1alpha2 "github.com/wso2/apk/common-go-libs/apis/cp/v1alpha2"
+	"github.com/wso2/apk/common-go-libs/constants"
 	"github.com/wso2/apk/common-go-libs/utils"
 	k8error "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -96,6 +97,11 @@ func (k8sArtifactDeployer K8sArtifactDeployer) UpdateApplication(application ser
 	return nil
 }
 
+// UpdateKeyMappings updates a key mapping
+func (k8sArtifactDeployer K8sArtifactDeployer) UpdateKeyMappings(keyMapping server.ApplicationKeyMapping) error {
+	return nil
+}
+
 // DeploySubscription deploys a subscription
 func (k8sArtifactDeployer K8sArtifactDeployer) DeploySubscription(subscription server.Subscription) error {
 	crSubscription := cpv1alpha2.Subscription{ObjectMeta: v1.ObjectMeta{Name: subscription.UUID, Namespace: utils.GetOperatorPodNamespace()},
@@ -151,7 +157,7 @@ func (k8sArtifactDeployer K8sArtifactDeployer) DeployKeyMappings(keyMapping serv
 	if crApplication.Spec.SecuritySchemes != nil {
 		securitySchemes = *crApplication.Spec.SecuritySchemes
 	}
-	if keyMapping.SecurityScheme == "OAuth2" {
+	if keyMapping.SecurityScheme == constants.OAuth2 {
 		if securitySchemes.OAuth2 == nil {
 			securitySchemes.OAuth2 = &cpv1alpha2.SecurityScheme{Environments: []cpv1alpha2.Environment{generateSecurityScheme(keyMapping)}}
 		} else {
@@ -257,7 +263,7 @@ func (k8sArtifactDeployer K8sArtifactDeployer) DeleteKeyMappings(keyMapping serv
 	}
 	if crApplication.Spec.SecuritySchemes != nil {
 		securitySchemes := *crApplication.Spec.SecuritySchemes
-		if keyMapping.SecurityScheme == "OAuth2" && securitySchemes.OAuth2 != nil {
+		if keyMapping.SecurityScheme == constants.OAuth2 && securitySchemes.OAuth2 != nil {
 			if securitySchemes.OAuth2.Environments != nil && len(securitySchemes.OAuth2.Environments) > 0 {
 				environments := make([]cpv1alpha2.Environment, 0)
 				for _, environment := range securitySchemes.OAuth2.Environments {