feat: add devlake-config update method

Signed-off-by: Yingchu Chen <yingchu.chen@merico.dev>

Author: warren830

internal/pkg/plugin/devlakeconfig/create.go

@@ -10,7 +10,7 @@ func Create(options map[string]interface{}) (map[string]interface{}, error) {
 	operator := &plugininstaller.Operator{
 		PreExecuteOperations: plugininstaller.PreExecuteOperations{
 			validate,
-			RenderAuthConfig,
+			//RenderAuthConfig,
 		},
 		ExecuteOperations: plugininstaller.ExecuteOperations{
 			ApplyConfig,

internal/pkg/plugin/devlakeconfig/delete.go

@@ -9,7 +9,7 @@ func Delete(options map[string]interface{}) (bool, error) {
 	operator := &plugininstaller.Operator{
 		PreExecuteOperations: plugininstaller.PreExecuteOperations{
 			validate,
-			RenderAuthConfig,
+			//RenderAuthConfig,
 		},
 		ExecuteOperations: plugininstaller.ExecuteOperations{
 			DeleteConfig,

internal/pkg/plugin/devlakeconfig/devlakeconfig.go

@@ -4,10 +4,13 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 	"time"
 
+	"github.com/devstream-io/devstream/internal/pkg/configmanager"
+
 	"github.com/devstream-io/devstream/internal/pkg/plugininstaller"
 	"github.com/devstream-io/devstream/internal/pkg/statemanager"
 	"github.com/devstream-io/devstream/pkg/util/log"
@@ -17,24 +20,24 @@ var httpClient = &http.Client{
 	Timeout: 5 * time.Second,
 }
 
-func RenderAuthConfig(options plugininstaller.RawOptions) (plugininstaller.RawOptions, error) {
-	opts, err := NewOptions(options)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, p := range opts.Plugins {
-		for _, c := range p.Connections {
-			c.Token = c.Authx.Token
-			c.Username = c.Authx.Username
-			c.Password = c.Authx.Password
-			c.AppId = c.Authx.AppId
-			c.SecretKey = c.Authx.SecretKey
-		}
-	}
-
-	return opts.Encode()
-}
+//func RenderAuthConfig(options plugininstaller.RawOptions) (plugininstaller.RawOptions, error) {
+//	opts, err := NewOptions(options)
+//	if err != nil {
+//		return nil, err
+//	}
+//
+//	for _, p := range opts.Plugins {
+//		for _, c := range p.Connections {
+//			c.Auth.Token = c.Authx.Token
+//			c.Username = c.Authx.Username
+//			c.Password = c.Authx.Password
+//			c.AppId = c.Authx.AppId
+//			c.SecretKey = c.Authx.SecretKey
+//		}
+//	}
+//
+//	return opts.Encode()
+//}
 
 func ApplyConfig(options plugininstaller.RawOptions) error {
 	opts, err := NewOptions(options)
@@ -55,26 +58,108 @@ func ApplyConfig(options plugininstaller.RawOptions) error {
 func createConnections(host string, pluginName string, connections []Connection) error {
 	for i, c := range connections {
 		log.Infof("Connection %d: %s", i, c.Name)
-		configs, err := json.Marshal(c)
-		if err != nil {
+		if err := createConnection(host, pluginName, c); err != nil {
 			return err
 		}
-		log.Debugf("Connection configs: %s", string(configs))
+	}
+	log.Infof("All %s connections have been created.", pluginName)
+	return nil
+}
 
-		url := fmt.Sprintf("%s/plugins/%s/connections", strings.TrimRight(host, "/"), pluginName)
-		log.Debugf("URL: %s", url)
+func createConnection(host string, pluginName string, c Connection) error {
+	configs, err := json.Marshal(c)
+	if err != nil {
+		return err
+	}
+	fmt.Println(string(configs))
+	log.Debugf("Connection configs: %s", string(configs))
+	url := fmt.Sprintf("%s/plugins/%s/connections", strings.TrimRight(host, "/"), pluginName)
+	log.Debugf("URL: %s", url)
+	if err = apiClient(http.MethodPost, url, configs); err != nil {
+		return err
+	}
+	return nil
+}
 
-		if err := createConnection(url, configs); err != nil {
-			return err
-		}
+func updateConnection(host string, pluginName string, c Connection) error {
+	id, err := getConnection(host, pluginName, c)
+	if err != nil {
+		return err
+	}
+	if id == 0 {
+		return nil
+	}
+	configs, err := json.Marshal(c)
+	if err != nil {
+		return err
+	}
+	log.Debugf("UPDATE Connection configs: %s", string(configs))
+	url := fmt.Sprintf("%s/plugins/%s/connections/%d", strings.TrimRight(host, "/"), pluginName, id)
+	log.Debugf("URL: %s", url)
+	if err = apiClient(http.MethodPatch, url, configs); err != nil {
+		return err
 	}
+	return nil
+}
 
-	log.Infof("All %s connections have been created.", pluginName)
+func deleteConnection(host string, pluginName string, c Connection) error {
+	id, err := getConnection(host, pluginName, c)
+	if err != nil {
+		return err
+	}
+	if id == 0 {
+		return nil
+	}
+	configs, err := json.Marshal(c)
+	if err != nil {
+		return err
+	}
+	log.Debugf("DELETE Connection configs: %s", string(configs))
+	url := fmt.Sprintf("%s/plugins/%s/connections/%d", strings.TrimRight(host, "/"), pluginName, id)
+	log.Debugf("URL: %s", url)
+	if err = apiClient(http.MethodDelete, url, configs); err != nil {
+		return err
+	}
 	return nil
 }
 
-func createConnection(url string, bodyWithJson []byte) error {
-	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(bodyWithJson))
+func getConnection(host string, pluginName string, c Connection) (uint64, error) {
+	url := fmt.Sprintf("%s/plugins/%s/connections", strings.TrimRight(host, "/"), pluginName)
+	log.Debugf("URL: %s", url)
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		return 0, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return 0, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusOK {
+		resBody, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return 0, err
+		}
+		connections := make([]Connection, 0)
+		err = json.Unmarshal(resBody, &connections)
+		if err != nil {
+			return 0, err
+		}
+		for _, v := range connections {
+			if v.Name == c.Name {
+				return v.ID, nil
+			}
+		}
+		return 0, nil
+	}
+
+	return 0, fmt.Errorf(resp.Status)
+}
+
+func apiClient(method string, url string, bodyWithJson []byte) error {
+	req, err := http.NewRequest(method, url, bytes.NewBuffer(bodyWithJson))
 	if err != nil {
 		return err
 	}
@@ -99,7 +184,66 @@ func DeleteConfig(options plugininstaller.RawOptions) error {
 }
 
 func UpdateConfig(options plugininstaller.RawOptions) error {
-	// TODO(daniel-hutao): implement later
+	opts, err := NewOptions(options)
+	if err != nil {
+		return err
+	}
+	cfg, err := configmanager.NewManager("config-devlake-config.yaml").LoadConfig()
+	if err != nil {
+		return err
+	}
+	smgr, _ := statemanager.NewManager(*cfg.State)
+	stateLakeConfig := smgr.GetState("devlake-config_default")
+	// map connections to pluginName
+	pluginConnections := make(map[string][]Connection)
+	states, err := NewOptions(stateLakeConfig.Options)
+	if err != nil {
+		return err
+	}
+
+	for _, v := range states.Plugins {
+		pluginConnections[v.Name] = v.Connections
+	}
+
+	for _, p := range opts.Plugins {
+		// Map Connection.Name -> Connection for config
+		configConnectionMap := make(map[string]Connection)
+		// Map Connection.Name -> Connection for state
+		stateConnectionMap := make(map[string]Connection)
+		for _, configConnection := range p.Connections {
+			configConnectionMap[configConnection.Name] = configConnection
+		}
+		// Construct a map
+		for _, stateConnection := range pluginConnections[p.Name] {
+			stateConnectionMap[stateConnection.Name] = stateConnection
+		}
+		for k := range configConnectionMap {
+			// Create connection which is not in State
+			if _, ok := stateConnectionMap[k]; !ok {
+				if err = createConnection(opts.DevLakeAddr, p.Name, configConnectionMap[k]); err != nil {
+					return err
+				}
+				continue
+			}
+			// Update connection which is different from State
+			if stateConnectionMap[k] != configConnectionMap[k] {
+				if err = updateConnection(opts.DevLakeAddr, p.Name, configConnectionMap[k]); err != nil {
+					return err
+				}
+				continue
+			}
+		}
+		for k := range stateConnectionMap {
+			// Delete connection which is not in config
+			if _, ok := configConnectionMap[k]; !ok {
+				if err = deleteConnection(opts.DevLakeAddr, p.Name, stateConnectionMap[k]); err != nil {
+					return err
+				}
+				continue
+			}
+		}
+	}
+
 	return nil
 }
 

internal/pkg/plugin/devlakeconfig/options.go

@@ -43,8 +43,8 @@ type DevLakePlugin struct {
 
 type Connection struct {
 	staging.RestConnection `mapstructure:",squash"`
-	Authx                  Auth `mapstructure:"auth" validate:"required"`
-	Auth                   `mapstructure:",squash"`
+	//Authx                  Auth `mapstructure:"auth" validate:"required"`
+	Auth `mapstructure:",squash"`
 }
 
 type Auth struct {

internal/pkg/plugin/devlakeconfig/read.go

@@ -10,7 +10,7 @@ func Read(options map[string]interface{}) (map[string]interface{}, error) {
 	operator := &plugininstaller.Operator{
 		PreExecuteOperations: plugininstaller.PreExecuteOperations{
 			validate,
-			RenderAuthConfig,
+			//RenderAuthConfig,
 		},
 		GetStateOperation: GetState,
 	}

internal/pkg/plugin/devlakeconfig/update.go

@@ -10,7 +10,7 @@ func Update(options map[string]interface{}) (map[string]interface{}, error) {
 	operator := &plugininstaller.Operator{
 		PreExecuteOperations: plugininstaller.PreExecuteOperations{
 			validate,
-			RenderAuthConfig,
+			//RenderAuthConfig,
 		},
 		ExecuteOperations: plugininstaller.ExecuteOperations{
 			UpdateConfig,

pkg/util/downloader/resource_test.go

@@ -9,8 +9,9 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
-	"github.com/devstream-io/devstream/pkg/util/downloader"
 	"github.com/onsi/gomega/ghttp"
+
+	"github.com/devstream-io/devstream/pkg/util/downloader"
 )
 
 var _ = Describe("ResourceClient struct", func() {