feat: Runtime variables and system config change reload

go.sum

@@ -730,6 +730,7 @@ k8s.io/apimachinery v0.0.0-20191219145857-f69eda767ee8/go.mod h1:mhhO3hoLkWO+2eC
 k8s.io/apimachinery v0.16.4/go.mod h1:llRdnznGEAqC3DcNm6yEj472xaFVfLM7hnYofMb12tQ=
 k8s.io/apimachinery v0.16.7-beta.0 h1:1cNiN7ZXJzlWq7dnWojG5UcrX1AIfQqpbyuzhu7Bhsc=
 k8s.io/apimachinery v0.16.7-beta.0/go.mod h1:mhhO3hoLkWO+2eCvqjPtH2Ly92l9nJDwsswzWKpkN2w=
+k8s.io/apimachinery v0.18.3 h1:pOGcbVAhxADgUYnjS08EFXs9QMl8qaH5U4fr5LGUrSk=
 k8s.io/client-go v0.0.0-20191225075139-73fd2ddc9180/go.mod h1:ksVkYlACXo9hR9AV+cYyCkuWL1xnWcGtAFxsfqMcozg=
 k8s.io/client-go v0.16.4 h1:sf+FEZXYhJNjpTZapQDLvvN+0kBeUTxCYxlXcVdhv2E=
 k8s.io/client-go v0.16.4/go.mod h1:ZgxhFDxSnoKY0J0U2/Y1C8obKDdlhGPZwA7oHH863Ok=

main.go

@@ -5,6 +5,12 @@ import (
 	"flag"
 	"fmt"
 	_ "github.com/onepanelio/core/db"
+	corev1 "k8s.io/api/core/v1"
+	apiv1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	k8runtime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/tools/cache"
 	"net"
 	"net/http"
 
@@ -37,29 +43,44 @@ var (
 func main() {
 	flag.Parse()
 
-	kubeConfig := v1.NewConfig()
-	client, err := v1.NewClient(kubeConfig, nil, nil)
-	if err != nil {
-		log.Fatalf("Failed to connect to Kubernetes cluster: %v", err)
-	}
-	sysConfig, err := client.GetSystemConfig()
-	if err != nil {
-		log.Fatalf("Failed to get system config: %v", err)
-	}
+	// stopCh is used to indicate when the RPC server should reload.
+	// We do this when the configuration has been changed, so the server has the latest configuration
+	stopCh := make(chan struct{})
 
-	databaseDataSourceName := fmt.Sprintf("host=%v user=%v password=%v dbname=%v sslmode=disable",
-		sysConfig["databaseHost"], sysConfig["databaseUsername"], sysConfig["databasePassword"], sysConfig["databaseName"])
+	go func() {
+		for {
+			kubeConfig := v1.NewConfig()
+			client, err := v1.NewClient(kubeConfig, nil, nil)
+			if err != nil {
+				log.Fatalf("Failed to connect to Kubernetes cluster: %v", err)
+			}
 
-	db := sqlx.MustConnect(sysConfig["databaseDriverName"], databaseDataSourceName)
-	if err := goose.Run("up", db.DB, "db"); err != nil {
-		log.Fatalf("Failed to run database migrations: %v", err)
-	}
+			client.ClearSystemConfigCache()
+			sysConfig, err := client.GetSystemConfig()
+			if err != nil {
+				log.Fatalf("Failed to get system config: %v", err)
+			}
+
+			databaseDataSourceName := fmt.Sprintf("host=%v user=%v password=%v dbname=%v sslmode=disable",
+				sysConfig["databaseHost"], sysConfig["databaseUsername"], sysConfig["databasePassword"], sysConfig["databaseName"])
+
+			db := sqlx.MustConnect(sysConfig["databaseDriverName"], databaseDataSourceName)
+			if err := goose.Run("up", db.DB, "db"); err != nil {
+				log.Fatalf("Failed to run database migrations: %v", err)
+			}
+
+			s := startRPCServer(db, kubeConfig, sysConfig, stopCh)
+
+			<-stopCh
+
+			s.Stop()
+		}
+	}()
 
-	go startRPCServer(db, kubeConfig, sysConfig)
 	startHTTPProxy()
 }
 
-func startRPCServer(db *v1.DB, kubeConfig *v1.Config, sysConfig v1.SystemConfig) {
+func startRPCServer(db *v1.DB, kubeConfig *v1.Config, sysConfig v1.SystemConfig, stopCh chan struct{}) *grpc.Server {
 	log.Printf("Starting RPC server on port %v", *rpcPort)
 	lis, err := net.Listen("tcp", *rpcPort)
 	if err != nil {
@@ -103,9 +124,25 @@ func startRPCServer(db *v1.DB, kubeConfig *v1.Config, sysConfig v1.SystemConfig)
 	api.RegisterWorkspaceTemplateServiceServer(s, server.NewWorkspaceTemplateServer())
 	api.RegisterWorkspaceServiceServer(s, server.NewWorkspaceServer())
 
-	if err := s.Serve(lis); err != nil {
-		log.Fatalf("Failed to serve RPC server: %v", err)
+	client, err := v1.NewClient(kubeConfig, db, sysConfig)
+	if err != nil {
+		log.Fatalf("Unable to create client: %v", err)
 	}
+
+	go watchConfigmapChanges(client, "onepanel", stopCh, func(configMap *corev1.ConfigMap) error {
+		log.Printf("Configmap changed")
+		stopCh <- struct{}{}
+
+		return nil
+	})
+
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("Failed to serve RPC server: %v", err)
+		}
+	}()
+
+	return s
 }
 
 func startHTTPProxy() {
@@ -159,3 +196,49 @@ func registerHandler(register registerFunc, ctx context.Context, mux *runtime.Se
 		log.Fatalf("Failed to register handler: %v", err)
 	}
 }
+
+// watchConfigmapChanges sets up a listener for configmap changes and calls the onChange function when it happens
+func watchConfigmapChanges(client *v1.Client, namespace string, stopCh <-chan struct{}, onChange func(*corev1.ConfigMap) error) {
+	restClient := client.CoreV1().RESTClient()
+	resource := "configmaps"
+	fieldSelector := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", "onepanel"))
+	listFunc := func(options apiv1.ListOptions) (k8runtime.Object, error) {
+		options.FieldSelector = fieldSelector.String()
+		req := restClient.Get().
+			Namespace(namespace).
+			Resource(resource).
+			VersionedParams(&options, apiv1.ParameterCodec)
+		return req.Do().Get()
+	}
+	watchFunc := func(options apiv1.ListOptions) (watch.Interface, error) {
+		options.Watch = true
+		options.FieldSelector = fieldSelector.String()
+		req := restClient.Get().
+			Namespace(namespace).
+			Resource(resource).
+			VersionedParams(&options, apiv1.ParameterCodec)
+		return req.Watch()
+	}
+	source := &cache.ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
+	_, controller := cache.NewInformer(
+		source,
+		&corev1.ConfigMap{},
+		0,
+		cache.ResourceEventHandlerFuncs{
+			UpdateFunc: func(old, new interface{}) {
+				oldCM := old.(*corev1.ConfigMap)
+				newCM := new.(*corev1.ConfigMap)
+				if oldCM.ResourceVersion == newCM.ResourceVersion {
+					return
+				}
+				if newCm, ok := new.(*corev1.ConfigMap); ok {
+					log.Infof("Detected ConfigMap update.")
+					if err := onChange(newCm); err != nil {
+						log.Errorf("Error on calling onChange callback: %v", err)
+					}
+				}
+			},
+		})
+	controller.Run(stopCh)
+	log.Infof("Watching config map updates")
+}

pkg/config.go

@@ -26,6 +26,12 @@ func (c *Client) getConfigMap(namespace, name string) (configMap *ConfigMap, err
 	return
 }
 
+// ClearSystemConfigCache wipes out the cached system configuration so that the next call to
+// GetSystemConfig will pull it from the resources
+func (c *Client) ClearSystemConfigCache() {
+	c.systemConfig = nil
+}
+
 // GetSystemConfig loads various system configurations and bundles them into a map.
 // The configuration is cached once it is loaded, and that cached value is used from here on out.
 func (c *Client) GetSystemConfig() (config map[string]string, err error) {

pkg/workflow_execution.go

@@ -6,11 +6,11 @@ import (
 	"errors"
 	"fmt"
 	sq "github.com/Masterminds/squirrel"
-	"github.com/ghodss/yaml"
 	"github.com/onepanelio/core/pkg/util/label"
 	"github.com/onepanelio/core/pkg/util/pagination"
 	"github.com/onepanelio/core/pkg/util/ptr"
 	uid2 "github.com/onepanelio/core/pkg/util/uid"
+	"gopkg.in/yaml.v2"
 	"io"
 	"io/ioutil"
 	"k8s.io/apimachinery/pkg/watch"
@@ -328,7 +328,9 @@ func (c *Client) ValidateWorkflowExecution(namespace string, manifest []byte) (e
 	return
 }
 
-func (c *Client) CreateWorkflowExecution(namespace string, workflow *WorkflowExecution) (*WorkflowExecution, error) {
+// CreateWorkflowExecution creates an argo workflow execution and related resources.
+// Note that the workflow template is loaded from the database/k8s, so workflow.WorkflowTemplate.Manifest is not used.
+func (c *Client) CreateWorkflowExecution(namespace string, workflow *WorkflowExecution, runtimeVars *RuntimeVars) (*WorkflowExecution, error) {
 	workflowTemplate, err := c.GetWorkflowTemplate(namespace, workflow.WorkflowTemplate.UID, workflow.WorkflowTemplate.Version)
 	if err != nil {
 		log.WithFields(log.Fields{
@@ -339,6 +341,31 @@ func (c *Client) CreateWorkflowExecution(namespace string, workflow *WorkflowExe
 		return nil, util.NewUserError(codes.NotFound, "Error with getting workflow template.")
 	}
 
+	if runtimeVars != nil && workflowTemplate.ArgoWorkflowTemplate != nil {
+		argoTemplate := workflowTemplate.ArgoWorkflowTemplate
+		for _, param := range runtimeVars.AdditionalParameters {
+			argoTemplate.Spec.Arguments.Parameters = append(argoTemplate.Spec.Arguments.Parameters, wfv1.Parameter{
+				Name:  param.Name,
+				Value: param.Value,
+			})
+		}
+
+		finalTemplates := make([]wfv1.Template, 0)
+		for i := range argoTemplate.Spec.Templates {
+			template := &argoTemplate.Spec.Templates[i]
+
+			if template.Name == "stateful-set-resource" {
+				template.Resource.Manifest = runtimeVars.StatefulSetManifest
+			}
+
+			if template.Name != runtimeVars.VirtualService.Name {
+				finalTemplates = append(finalTemplates, *template)
+			}
+		}
+		finalTemplates = append(finalTemplates, *runtimeVars.VirtualService)
+		workflowTemplate.ArgoWorkflowTemplate.Spec.Templates = finalTemplates
+	}
+
 	// TODO: Need to pull system parameters from k8s config/secret here, example: HOST
 	opts := &WorkflowExecutionOptions{
 		Labels:     &map[string]string{},
@@ -416,7 +443,7 @@ func (c *Client) CloneWorkflowExecution(namespace, uid string) (*WorkflowExecuti
 		return nil, err
 	}
 
-	return c.CreateWorkflowExecution(namespace, workflowExecution)
+	return c.CreateWorkflowExecution(namespace, workflowExecution, nil)
 }
 
 func (c *Client) insertPreWorkflowExecutionStatistic(namespace, name string, workflowTemplateVersionId uint64, createdAt time.Time, uid string, parameters []Parameter) (newId uint64, err error) {

pkg/workspace.go

@@ -98,10 +98,21 @@ func injectWorkspaceSystemParameters(namespace string, workspace *Workspace, wor
 }
 
 func (c *Client) createWorkspace(namespace string, parameters []byte, workspace *Workspace) (*Workspace, error) {
-	_, err := c.CreateWorkflowExecution(namespace, &WorkflowExecution{
+	systemConfig, err := c.GetSystemConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	runtimeVars, err := workspace.WorkspaceTemplate.RuntimeVars(systemConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = c.CreateWorkflowExecution(namespace, &WorkflowExecution{
 		Parameters:       workspace.Parameters,
 		WorkflowTemplate: workspace.WorkspaceTemplate.WorkflowTemplate,
-	})
+	}, runtimeVars)
+
 	if err != nil {
 		return nil, err
 	}
@@ -116,7 +127,6 @@ func (c *Client) createWorkspace(namespace string, parameters []byte, workspace
 			"started_at":                 time.Now().UTC(),
 			"workspace_template_id":      workspace.WorkspaceTemplate.ID,
 			"workspace_template_version": workspace.WorkspaceTemplate.Version,
-			"url":                        workspace.URL,
 		}).
 		Suffix("RETURNING id, created_at").
 		RunWith(c.DB).
@@ -154,7 +164,6 @@ func (c *Client) CreateWorkspace(namespace string, workspace *Workspace) (*Works
 	if sysHost == nil {
 		return nil, fmt.Errorf("sys-host parameter not found")
 	}
-	workspace.URL = *sysHost
 
 	existingWorkspace, err := c.GetWorkspace(namespace, workspace.UID)
 	if err != nil {
@@ -394,7 +403,7 @@ func (c *Client) updateWorkspace(namespace, uid, workspaceAction, resourceAction
 	_, err = c.CreateWorkflowExecution(namespace, &WorkflowExecution{
 		Parameters:       workspace.Parameters,
 		WorkflowTemplate: workspace.WorkspaceTemplate.WorkflowTemplate,
-	})
+	}, nil)
 	if err != nil {
 		return
 	}