fix: Add Multi Module Support (#8)

* fix: test improvments

* add tests for dgate cli and dgclient

* refactor dgate cli run function, add more test to increase coverage for dgate-cli

* simplify code and add more test for better coverage

* fix: nil ref error

* fix nil ref and remove WithHttpClient hook for dgclient

* small refactor on dgclient for better support for changing the client

* fix redirect bug

* implement multi module and add multi module functionality

* fix: change log compact function and setupModule state bug

* fix logic issue

* fix benchmark test

* revert changes

functional-tests/admin_tests/modify_response.ts

@@ -4,9 +4,9 @@ export const responseModifier = async (ctx) => {
             const uuidData = await res.json();
             console.log("INFO uuid", JSON.stringify(uuidData));
             const resp = ctx.upstream();
-            const results = await resp.getJson();
+            const results = await resp.readJson();
             results.data.uuid = uuidData.uuid;
-            return resp.setStatus(200).setJson(results);
+            return resp.status(200).writeJson(results);
         });
 };
 

functional-tests/admin_tests/modify_response_test.sh

@@ -37,6 +37,6 @@ dgate-cli route create \
     namespace=test-ns \
     service='base_svc'
 
-curl ${PROXY_URL}/test -H Host:test.com
+curl -s ${PROXY_URL}/test -H Host:test.com
 
 echo "Modify Response Test Passed"

functional-tests/admin_tests/multi_module_test.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+
+set -eo xtrace
+
+ADMIN_URL=${ADMIN_URL:-"http://localhost:9080"}
+PROXY_URL=${PROXY_URL:-"http://localhost"}
+
+DIR="$( cd "$( dirname "$0" )" && pwd )"
+
+export DGATE_ADMIN_API=$ADMIN_URL
+
+dgate-cli namespace create \
+    name=multimod-test-ns
+
+dgate-cli domain create \
+    name=multimod-test-dm \
+    patterns:='["multimod-test.com"]' \
+    namespace=multimod-test-ns
+
+MOD_B64=$(base64 <<-END
+export {
+    requestModifier,
+} from './multimod2';
+import {
+    responseModifier as resMod,
+} from './multimod2';
+const responseModifier = async (ctx) => {
+    console.log('responseModifier executed from multimod1')
+    return resMod(ctx);
+};
+END
+
+)
+
+dgate-cli module create \
+    name=multimod1 \
+    payload="$MOD_B64" \
+    namespace=multimod-test-ns
+
+MOD_B64=$(base64 <<-END
+const reqMod = (ctx) => ctx.request().writeJson({a: 1});
+const resMod = async (ctx) => ctx.upstream()?.writeJson({
+    upstream_body: await ctx.upstream()?.readJson(),
+    upstream_headers: ctx.upstream()?.headers,
+    upsteam_status: ctx.upstream()?.statusCode,
+    upstream_statusText: ctx.upstream()?.statusText,
+});
+export {
+    reqMod as requestModifier,
+    resMod as responseModifier,
+};
+END
+
+)
+
+dgate-cli module create name=multimod2 \
+    payload="$MOD_B64" namespace=multimod-test-ns
+
+URL='http://localhost:8888'
+dgate-cli service create name=base_svc \
+    urls="$URL/a","$URL/b","$URL/c" \
+    namespace=multimod-test-ns
+
+dgate-cli route create name=base_rt \
+    paths=/,/multimod-test \
+    methods:='["GET"]' \
+    modules=multimod1,multimod2 \
+    service=base_svc \
+    stripPath:=true \
+    preserveHost:=true \
+    namespace=multimod-test-ns
+
+
+curl -s --fail-with-body ${PROXY_URL}/ -H Host:multimod-test.com
+curl -s --fail-with-body ${PROXY_URL}/multimod-test -H Host:multimod-test.com
+
+echo "Multi Module Test Passed"

internal/admin/admin_api.go

@@ -2,6 +2,7 @@ package admin
 
 import (
 	"fmt"
+	"io"
 	"log"
 	"net/http"
 	"os"
@@ -89,8 +90,11 @@ func StartAdminAPI(conf *config.DGateConfig, proxyState *proxy.ProxyState) {
 					}
 					respMap["method"] = r.Method
 					respMap["path"] = r.URL.String()
-					respMap["remote_addr"] = r.RemoteAddr
+					if body, err := io.ReadAll(r.Body); err == nil {
+						respMap["body"] = string(body)
+					}
 					respMap["host"] = r.Host
+					respMap["remote_addr"] = r.RemoteAddr
 					respMap["req_headers"] = r.Header
 					if conf.TestServerConfig.EnableEnvVars {
 						respMap["env"] = os.Environ()

internal/admin/admin_raft.go

@@ -77,25 +77,25 @@ func setupRaft(conf *config.DGateConfig, server *chi.Mux, ps *proxy.ProxyState)
 		panic(fmt.Errorf("invalid scheme: %s", adminConfig.Replication.AdvertScheme))
 	}
 
-	trans := rafthttp.NewHTTPTransport(address,
-		http.DefaultClient, raftHttpLogger,
-		adminConfig.Replication.AdvertScheme+
-			"://(address)/raft",
+	transport := rafthttp.NewHTTPTransport(
+		address, http.DefaultClient, raftHttpLogger,
+		adminConfig.Replication.AdvertScheme+"://(address)/raft",
+	)
+	raftNode, err := raft.NewRaft(
+		raftConfig, newDGateAdminFSM(ps),
+		lstore, sstore, snapstore, transport,
 	)
-	raftNode, err := raft.NewRaft(raftConfig, newDGateAdminFSM(ps),
-		lstore, sstore, snapstore, trans)
 	if err != nil {
 		panic(err)
 	}
 
 	ps.SetupRaft(raftNode, raftConfig)
 	// Setup raft handler
-	server.Handle("/raft/*", trans)
+	server.Handle("/raft/*", transport)
 
 	raftAdminLogger := ps.Logger().With().Str("component", "raftAdmin").Logger()
 	raftAdmin := raftadmin.NewRaftAdminHTTPServer(
-		raftNode, raftAdminLogger,
-		[]raft.ServerAddress{address},
+		raftNode, raftAdminLogger, []raft.ServerAddress{address},
 	)
 
 	// Setup handler raft

internal/admin/routes/module_routes.go

@@ -120,6 +120,6 @@ func ConfigureModuleAPI(server chi.Router, proxyState *proxy.ProxyState, appConf
 			util.JsonError(w, http.StatusNotFound, "module not found")
 			return
 		}
-		util.JsonResponse(w, http.StatusOK, mod)
+		util.JsonResponse(w, http.StatusOK, spec.TransformDGateModule(mod))
 	})
 }

internal/proxy/change_log.go

@@ -14,11 +14,11 @@ import (
 )
 
 // processChangeLog - processes a change log and applies the change to the proxy state
-func (ps *ProxyState) processChangeLog(
-	cl *spec.ChangeLog, reload, store bool,
-) (err error) {
+func (ps *ProxyState) processChangeLog(cl *spec.ChangeLog, reload, store bool) (err error) {
 	if cl == nil {
-		cl = &spec.ChangeLog{Cmd: spec.NoopCommand}
+		cl = &spec.ChangeLog{
+			Cmd: spec.NoopCommand,
+		}
 	} else if !cl.Cmd.IsNoop() {
 		switch cl.Cmd.Resource() {
 		case spec.Namespaces:
@@ -81,7 +81,7 @@ func (ps *ProxyState) processChangeLog(
 			err = fmt.Errorf("unknown command: %s", cl.Cmd)
 		}
 		if err != nil {
-			ps.logger.Err(err).Msg("error processing change log")
+			ps.logger.Err(err).Msg("decoding or processing change log")
 			return
 		}
 	}
@@ -273,11 +273,7 @@ func (ps *ProxyState) applyChange(changeLog *spec.ChangeLog) <-chan error {
 	return done
 }
 
-func (ps *ProxyState) rollbackChange(changeLog *spec.ChangeLog) {
-	panic("not implemented")
-}
-
-func (ps *ProxyState) restoreFromChangeLogs() error {
+func (ps *ProxyState) restoreFromChangeLogs(directApply bool) error {
 	// restore state change logs
 	logs, err := ps.store.FetchChangeLogs()
 	if err != nil {
@@ -294,12 +290,23 @@ func (ps *ProxyState) restoreFromChangeLogs() error {
 				Interface("changeLog: "+cl.Name, cl).Msgf("restoring change log index: %d", i)
 			err = ps.processChangeLog(cl, false, false)
 			if err != nil {
+				if ps.config.Debug {
+					ps.logger.Err(err).
+						Str("namespace", cl.Namespace).
+						Msg("error restorng from change logs")
+					continue
+				}
 				return err
 			}
 		}
-
-		if err = ps.processChangeLog(nil, true, false); err != nil {
-			return err
+		if !directApply {
+			if err = ps.processChangeLog(nil, true, false); err != nil {
+				return err
+			}
+		} else {
+			if err = ps.reconfigureState(false, nil); err != nil {
+				return nil
+			}
 		}
 
 		// TODO: optionally compact change logs through a flag in config?
@@ -327,13 +334,11 @@ func (ps *ProxyState) compactChangeLogs(logs []*spec.ChangeLog) (int, error) {
 }
 
 /*
-compactChangeLogsRemoveList - compacts a list of change logs by removing redundant logs
-
-TODO: perhaps add flag for compacting change logs on startup (mark as experimental)
+compactChangeLogsRemoveList - compacts a list of change logs by removing redundant logs.
 
 compaction rules:
-- if an add command is followed by a delete command with matching keys, remove both commands
-- if an add command is followed by another add command with matching keys, remove the first add command
+  - if an add command is followed by a delete command with matching keys, remove both commands
+  - if an add command is followed by another add command with matching keys, remove the first add command
 */
 func compactChangeLogsRemoveList(logger *zerolog.Logger, logs []*spec.ChangeLog) []*spec.ChangeLog {
 	removeList := make([]*spec.ChangeLog, 0)
@@ -350,9 +355,11 @@ START:
 				logs = append(logs[:i-1], logs[i:]...)
 				goto START
 			}
+
 			commonResource := prevLog.Cmd.Resource() == curLog.Cmd.Resource()
 			if prevLog.Cmd.Action() == spec.Add && curLog.Cmd.Action() == spec.Delete && commonResource {
-				// Rule 1: if an add command is followed by a delete command with matching keys, remove both commands
+				// Rule 1: if an add command is followed by a delete
+				//   command with matching keys, remove both commands
 				if prevLog.Name == curLog.Name && prevLog.Namespace == curLog.Namespace {
 					removeList = append(removeList, prevLog, curLog)
 					logs = append(logs[:i-1], logs[i+1:]...)
@@ -362,7 +369,8 @@ START:
 
 			commonAction := prevLog.Cmd.Action() == curLog.Cmd.Action()
 			if prevLog.Cmd.Action() == spec.Add && commonAction && commonResource {
-				// Rule 2: if an add command is followed by another add command with matching keys, remove the first add command
+				// Rule 2: if an add command is followed by another add
+				//   command with matching keys, remove the first add command
 				if prevLog.Name == curLog.Name && prevLog.Namespace == curLog.Namespace {
 					removeList = append(removeList, prevLog)
 					logs = append(logs[:i-1], logs[i:]...)