Skip to content

Commit

Permalink
Merge pull request #460 from kube-tarian/postgres-migrate
Browse files Browse the repository at this point in the history
migrate database from cassandra to postgres
  • Loading branch information
vramk23 authored May 19, 2024
2 parents ab4353d + ed2a8f9 commit a56409c
Show file tree
Hide file tree
Showing 170 changed files with 3,782 additions and 6,568 deletions.
27 changes: 14 additions & 13 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,16 @@ DEPLOYMENT_WORKER_APP_NAME := deployment-worker
CONFIG_WORKER_APP_NAME := config-worker
BUILD := 0.1.1

.PHONY: gen-protoc
gen-protoc:
mkdir -p server/pkg/pb/serverpb
mkdir -p capten/agent/internal/pb/captenpluginspb
mkdir -p capten/common-pkg/pb/captenpluginspb
mkdir -p server/pkg/pb/agentpb
mkdir -p server/pkg/pb/captenpluginspb
mkdir -p server/pkg/pb/pluginstorepb
mkdir -p capten/common-pkg/vault-cred/vaultcredpb
mkdir -p capten/common-pkg/agentpb
mkdir -p capten/common-pkg/cluster-plugins/clusterpluginspb
mkdir -p capten/common-pkg/pb/vaultcredpb
mkdir -p capten/common-pkg/pb/agentpb
mkdir -p capten/common-pkg/pb/clusterpluginspb
mkdir -p server/pkg/pb/clusterpluginspb

cd proto && protoc --go_out=../server/pkg/pb/serverpb/ --go_opt=paths=source_relative \
Expand All @@ -23,29 +24,29 @@ gen-protoc:
cd proto && protoc --go_out=../server/pkg/pb/agentpb --go_opt=paths=source_relative \
--go-grpc_out=../server/pkg/pb/agentpb --go-grpc_opt=paths=source_relative \
./agent.proto

cd proto && protoc --go_out=../capten/common-pkg/agentpb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/agentpb --go-grpc_opt=paths=source_relative \
cd proto && protoc --go_out=../capten/common-pkg/pb/agentpb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/pb/agentpb --go-grpc_opt=paths=source_relative \
./agent.proto

cd proto && protoc --go_out=../server/pkg/pb/captenpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../server/pkg/pb/captenpluginspb --go-grpc_opt=paths=source_relative \
./capten_plugins.proto

cd proto && protoc --go_out=../capten/agent/internal/pb/captenpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../capten/agent/internal/pb/captenpluginspb --go-grpc_opt=paths=source_relative \
cd proto && protoc --go_out=../capten/common-pkg/pb/captenpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/pb/captenpluginspb --go-grpc_opt=paths=source_relative \
./capten_plugins.proto

cd proto && protoc --go_out=../capten/common-pkg/vault-cred/vaultcredpb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/vault-cred/vaultcredpb --go-grpc_opt=paths=source_relative \
cd proto && protoc --go_out=../capten/common-pkg/pb/vaultcredpb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/pb/vaultcredpb --go-grpc_opt=paths=source_relative \
./vault_cred.proto

cd proto && protoc --go_out=../server/pkg/pb/pluginstorepb --go_opt=paths=source_relative \
--go-grpc_out=../server/pkg/pb/pluginstorepb --go-grpc_opt=paths=source_relative \
./plugin_store.proto

cd proto && protoc --go_out=../capten/common-pkg/cluster-plugins/clusterpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/cluster-plugins/clusterpluginspb --go-grpc_opt=paths=source_relative \
cd proto && protoc --go_out=../capten/common-pkg/pb/clusterpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../capten/common-pkg/pb/clusterpluginspb --go-grpc_opt=paths=source_relative \
./cluster_plugins.proto
cd proto && protoc --go_out=../server/pkg/pb/clusterpluginspb --go_opt=paths=source_relative \
--go-grpc_out=../server/pkg/pb/clusterpluginspb --go-grpc_opt=paths=source_relative \
Expand Down
13 changes: 5 additions & 8 deletions capten/agent/internal/api/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,12 @@ import (
"fmt"

"github.com/intelops/go-common/logging"
captenstore "github.com/kube-tarian/kad/capten/agent/internal/capten-store"
"github.com/kube-tarian/kad/capten/agent/internal/config"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/agent/internal/temporalclient"
"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/cluster-plugins/clusterpluginspb"
pluginconfigtore "github.com/kube-tarian/kad/capten/common-pkg/pluginconfig-store"
captenstore "github.com/kube-tarian/kad/capten/common-pkg/capten-store"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/clusterpluginspb"
)

var _ agentpb.AgentServer = &Agent{}
Expand All @@ -22,13 +21,12 @@ type Agent struct {
clusterpluginspb.UnimplementedClusterPluginsServer
tc *temporalclient.Client
as *captenstore.Store
pas *pluginconfigtore.Store
log logging.Logger
cfg *config.SericeConfig
createPr bool
}

func NewAgent(log logging.Logger, cfg *config.SericeConfig, as *captenstore.Store, pas *pluginconfigtore.Store) (*Agent, error) {
func NewAgent(log logging.Logger, cfg *config.SericeConfig, as *captenstore.Store) (*Agent, error) {
var tc *temporalclient.Client
var err error

Expand All @@ -40,7 +38,6 @@ func NewAgent(log logging.Logger, cfg *config.SericeConfig, as *captenstore.Stor
agent := &Agent{
tc: tc,
as: as,
pas: pas,
cfg: cfg,
log: log,
}
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/agent_cluster_apps.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import (
"context"
"fmt"

"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
)

func (a *Agent) SyncApp(ctx context.Context, request *agentpb.SyncAppRequest) (
Expand Down
4 changes: 2 additions & 2 deletions capten/agent/internal/api/agent_cluster_apps_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import (
"testing"

"github.com/kube-tarian/kad/capten/agent/internal/config"
"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
"github.com/kube-tarian/kad/integrator/common-pkg/logging"
"github.com/stretchr/testify/suite"
"gopkg.in/yaml.v2"
Expand All @@ -27,7 +27,7 @@ func TestAgentTestSuite(t *testing.T) {
t.Fatal(err)
}*/

agent, err := NewAgent(agentSuite.logger, &config.SericeConfig{}, nil, nil)
agent, err := NewAgent(agentSuite.logger, &config.SericeConfig{}, nil)
if err != nil {
t.Fatal(err)
}
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/app_config_sso.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ import (
"context"

"github.com/kube-tarian/kad/capten/agent/internal/workers"
"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/credential"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
"github.com/kube-tarian/kad/capten/model"
"github.com/pkg/errors"
"gopkg.in/yaml.v2"
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/app_values_util.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@ import (
"strings"

"github.com/intelops/go-common/logging"
"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/credential"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
"github.com/kube-tarian/kad/capten/model"
"github.com/pkg/errors"
"gopkg.in/yaml.v2"
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/app_values_util_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import (
"testing"

"github.com/intelops/go-common/logging"
"github.com/kube-tarian/kad/capten/common-pkg/agentpb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/agentpb"
"github.com/stretchr/testify/require"
"gopkg.in/yaml.v2"
)
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/capten_plugin_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ package api
import (
"context"

"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
)

func (a *Agent) GetCaptenPlugins(ctx context.Context, request *captenpluginspb.GetCaptenPluginsRequest) (
Expand Down
42 changes: 18 additions & 24 deletions capten/agent/internal/api/cluster_plugin_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,14 @@ import (
"html/template"

"github.com/kube-tarian/kad/capten/agent/internal/workers"
"github.com/kube-tarian/kad/capten/common-pkg/cluster-plugins/clusterpluginspb"
pluginconfigstore "github.com/kube-tarian/kad/capten/common-pkg/pluginconfig-store"
"github.com/kube-tarian/kad/capten/common-pkg/pb/clusterpluginspb"
"github.com/kube-tarian/kad/capten/model"
"gopkg.in/yaml.v2"
)

func (a *Agent) GetClusterPlugins(ctx context.Context, request *clusterpluginspb.GetClusterPluginsRequest) (
*clusterpluginspb.GetClusterPluginsResponse, error) {
pluginConfigList, err := a.pas.GetAllPlugins()
pluginConfigList, err := a.as.GetAllClusterPluginConfigs()
if err != nil {
return &clusterpluginspb.GetClusterPluginsResponse{
Status: clusterpluginspb.StatusCode_INTERNRAL_ERROR,
Expand Down Expand Up @@ -44,31 +43,26 @@ func (a *Agent) DeployClusterPlugin(ctx context.Context, request *clusterplugins
*clusterpluginspb.DeployClusterPluginResponse, error) {
a.log.Infof("Recieved Plugin Deploy request for plugin %s, version %+v", request.Plugin.PluginName, request.Plugin.Version)

pluginConfig := &pluginconfigstore.PluginConfig{
Plugin: *request.Plugin,
}

values, err := replaceTemplateValuesInByteData(pluginConfig.Plugin.Values, pluginConfig.Plugin.OverrideValues)
values, err := replaceTemplateValuesInByteData(request.Plugin.Values, request.Plugin.OverrideValues)
if err != nil {
a.log.Errorf("failed to derive template values for plugin %s, %v", pluginConfig.PluginName, err)
a.log.Errorf("failed to derive template values for plugin %s, %v", request.Plugin.PluginName, err)
return &clusterpluginspb.DeployClusterPluginResponse{
Status: clusterpluginspb.StatusCode_INTERNRAL_ERROR,
StatusMessage: "failed to prepare plugin values",
}, nil
}

pluginConfig.Plugin.Values = values
pluginConfig.InstallStatus = string(model.AppIntallingStatus)
if err := a.pas.UpsertPluginConfig(pluginConfig); err != nil {
a.log.Errorf("failed to update plugin config data for plugin %s, %v", pluginConfig.PluginName, err)
request.Plugin.Values = values
request.Plugin.InstallStatus = string(model.AppIntallingStatus)
if err := a.as.UpsertClusterPluginConfig(request.Plugin); err != nil {
a.log.Errorf("failed to update plugin config data for plugin %s, %v", request.Plugin.PluginName, err)
return &clusterpluginspb.DeployClusterPluginResponse{
Status: clusterpluginspb.StatusCode_INTERNRAL_ERROR,
StatusMessage: "failed to update plugin config data",
}, nil
}

// deployReq := prepareAppDeployRequestFromPlugin(plugin)
go a.deployPluginWithWorkflow(request.Plugin, pluginConfig)
go a.deployPluginWithWorkflow(request.Plugin)

a.log.Infof("Triggerred plugin [%s] install", request.Plugin.PluginName)
return &clusterpluginspb.DeployClusterPluginResponse{
Expand All @@ -89,7 +83,7 @@ func (a *Agent) UnDeployClusterPlugin(ctx context.Context, request *clusterplugi
}, nil
}

pluginConfigdata, err := a.pas.GetPluginConfig(request.PluginName)
pluginConfigdata, err := a.as.GetClusterPluginConfig(request.PluginName)
if err != nil {
a.log.Errorf("failed to fetch plugin config record %s, %v", request.PluginName, err)
return &clusterpluginspb.UnDeployClusterPluginResponse{
Expand All @@ -99,7 +93,7 @@ func (a *Agent) UnDeployClusterPlugin(ctx context.Context, request *clusterplugi
}

pluginConfigdata.InstallStatus = string(model.AppUnInstallingStatus)
if err := a.pas.UpsertPluginConfig(pluginConfigdata); err != nil {
if err := a.as.UpsertClusterPluginConfig(pluginConfigdata); err != nil {
a.log.Errorf("failed to update plugin config status with UnInstalling for plugin %s, %v", pluginConfigdata.PluginName, err)
return &clusterpluginspb.UnDeployClusterPluginResponse{
Status: clusterpluginspb.StatusCode_INTERNRAL_ERROR,
Expand All @@ -116,7 +110,7 @@ func (a *Agent) UnDeployClusterPlugin(ctx context.Context, request *clusterplugi
}, nil
}

func (a *Agent) deployPluginWithWorkflow(plugin *clusterpluginspb.Plugin, pluginConfig *pluginconfigstore.PluginConfig) {
func (a *Agent) deployPluginWithWorkflow(plugin *clusterpluginspb.Plugin) {
wd := workers.NewDeployment(a.tc, a.log)
_, err := wd.SendEventV2(context.TODO(), wd.GetPluginWorkflowName(), string(model.AppInstallAction), plugin)
if err != nil {
Expand All @@ -125,23 +119,23 @@ func (a *Agent) deployPluginWithWorkflow(plugin *clusterpluginspb.Plugin, plugin
// a.log.Errorf("failed to update plugin config for plugin %s, %v", pluginConfig.PluginName, err)
// return
// }
a.log.Errorf("sendEventV2 failed, plugin: %s, reason: %v", pluginConfig.PluginName, err)
a.log.Errorf("sendEventV2 failed, plugin: %s, reason: %v", plugin.PluginName, err)
return
}
// TODO: workflow will update the final status
// Write a periodic scheduler which will go through all apps not in installed status and check the status till either success or failed.
// Make SendEventV2 asynchrounous so that periodic scheduler will take care of monitoring.
}

func (a *Agent) unInstallPluginWithWorkflow(request *clusterpluginspb.UnDeployClusterPluginRequest, pluginConfig *pluginconfigstore.PluginConfig) {
func (a *Agent) unInstallPluginWithWorkflow(request *clusterpluginspb.UnDeployClusterPluginRequest, plugin *clusterpluginspb.Plugin) {
wd := workers.NewDeployment(a.tc, a.log)
_, err := wd.SendDeleteEvent(context.TODO(), wd.GetPluginWorkflowName(), string(model.AppUnInstallAction), request)
if err != nil {
a.log.Errorf("failed to send delete event to workflow for plugin %s, %v", pluginConfig.PluginName, err)
a.log.Errorf("failed to send delete event to workflow for plugin %s, %v", request.PluginName, err)

pluginConfig.InstallStatus = string(model.AppUnUninstallFailedStatus)
if err := a.pas.UpsertPluginConfig(pluginConfig); err != nil {
a.log.Errorf("failed to update plugin config status with Installed for plugin %s, %v", pluginConfig.PluginName, err)
plugin.InstallStatus = string(model.AppUnUninstallFailedStatus)
if err := a.as.UpsertClusterPluginConfig(plugin); err != nil {
a.log.Errorf("failed to update plugin config status with Installed for plugin %s, %v", request.PluginName, err)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/container_registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import (

"github.com/google/uuid"
"github.com/intelops/go-common/credentials"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
)

const containerRegEntityName = "container-registry"
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/plugin_argocd_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import (
"strings"

"github.com/intelops/go-common/logging"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/plugins/argocd"
)

Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/plugin_cloud_provider_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import (

"github.com/google/uuid"
"github.com/intelops/go-common/credentials"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
)

const cloudProviderEntityName = "cloud-provider"
Expand Down
9 changes: 2 additions & 7 deletions capten/agent/internal/api/plugin_crossplane_project_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ import (
"context"
"fmt"

"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/agent/internal/workers"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/model"
)

Expand Down Expand Up @@ -168,7 +168,6 @@ func (a *Agent) configureCrossplaneGitRepo(req *model.CrossplaneProject, provide
&model.ConfigureParameters{Resource: crossplaneConfigUseCase, Action: model.CrossPlaneProjectSync}, ci)
if err != nil {
req.Status = string(model.CrossplaneProjectConfigurationFailed)
req.WorkflowId = "NA"
if err := a.as.UpsertCrossplaneProject(req); err != nil {
return fmt.Errorf("failed to update Cluster Gitopts Project, %v", err)
}
Expand All @@ -178,8 +177,6 @@ func (a *Agent) configureCrossplaneGitRepo(req *model.CrossplaneProject, provide
a.log.Infof("Crossplane Git project %s config workflow %s created", req.GitProjectUrl, wkfId)

req.Status = string(model.CrossplaneProjectConfigured)
req.WorkflowId = wkfId
req.WorkflowStatus = string(model.WorkFlowStatusStarted)
if err := a.as.UpsertCrossplaneProject(req); err != nil {
a.log.Errorf("failed to update Cluster Gitopts Project, %v", err)
return nil
Expand All @@ -193,10 +190,9 @@ func (a *Agent) configureCrossplaneGitRepo(req *model.CrossplaneProject, provide
func (a *Agent) monitorCrossplaneWorkflow(req *model.CrossplaneProject, wkfId string) {
// during system reboot start monitoring, add it in map or somewhere.
wd := workers.NewConfig(a.tc, a.log)
wkfResp, err := wd.GetWorkflowInformation(context.TODO(), wkfId)
_, err := wd.GetWorkflowInformation(context.TODO(), wkfId)
if err != nil {
req.Status = string(model.CrossplaneProjectConfigurationFailed)
req.WorkflowStatus = string(model.WorkFlowStatusFailed)
if err := a.as.UpsertCrossplaneProject(req); err != nil {
a.log.Errorf("failed to update Cluster Gitopts Project, %v", err)
return
Expand All @@ -206,7 +202,6 @@ func (a *Agent) monitorCrossplaneWorkflow(req *model.CrossplaneProject, wkfId st
}

req.Status = string(model.CrossplaneProjectConfigured)
req.WorkflowStatus = wkfResp.Status
if err := a.as.UpsertCrossplaneProject(req); err != nil {
a.log.Errorf("failed to update Cluster Gitopts Project, %v", err)
return
Expand Down
4 changes: 2 additions & 2 deletions capten/agent/internal/api/plugin_crossplane_provider_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import (
"context"

"github.com/google/uuid"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/model"
)

Expand Down Expand Up @@ -47,7 +47,7 @@ func (a *Agent) AddCrossplanProvider(ctx context.Context, request *captenplugins
Status: string(model.CrossPlaneProviderOutofSynch),
}

if err := a.as.InsertCrossplaneProvider(&provider); err != nil {
if err := a.as.UpsertCrossplaneProvider(&provider); err != nil {
a.log.Errorf("failed to store crossplane provider to DB, %v", err)
return &captenpluginspb.AddCrossplanProviderResponse{
Status: captenpluginspb.StatusCode_INTERNAL_ERROR,
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/plugin_git_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import (

"github.com/google/uuid"
"github.com/intelops/go-common/credentials"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
)

const gitProjectEntityName = "git-project"
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/plugin_managed_clusters_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ import (
"fmt"

"github.com/intelops/go-common/credentials"
"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
)

const ManagedClusterEntityName = "managedcluster"
Expand Down
2 changes: 1 addition & 1 deletion capten/agent/internal/api/plugin_tekton_apis.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@ package api
import (
"context"

"github.com/kube-tarian/kad/capten/agent/internal/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/agent/internal/workers"
"github.com/kube-tarian/kad/capten/common-pkg/pb/captenpluginspb"
"github.com/kube-tarian/kad/capten/model"
)

Expand Down
Loading

0 comments on commit a56409c

Please sign in to comment.