feat(agent): register target

distantmagic · May 13, 2024 · 622c7a7 · 622c7a7
1 parent f798766
commit 622c7a7
Show file tree

Hide file tree

Showing 16 changed files with 319 additions and 115 deletions.
diff --git a/agent/AgentConfiguration.go b/agent/AgentConfiguration.go
@@ -0,0 +1,11 @@
+package agent
+
+import "time"
+
+type AgentConfiguration struct {
+	ReportingIntervalMiliseconds uint
+}
+
+func (self *AgentConfiguration) GetReportingIntervalDuration() time.Duration {
+	return time.Duration(self.ReportingIntervalMiliseconds) * time.Millisecond
+}
diff --git a/agent/LlamaCppObserver.go b/agent/LlamaCppObserver.go
@@ -1,4 +1,46 @@
 package agent
 
+import (
+	"time"
+
+	"github.com/distantmagic/paddler/goroutine"
+	"github.com/distantmagic/paddler/llamacpp"
+	"github.com/distantmagic/paddler/management"
+	"github.com/hashicorp/go-hclog"
+)
+
 type LlamaCppObserver struct {
+	AgentConfiguration *AgentConfiguration
+	LlamaCppClient     *llamacpp.LlamaCppClient
+	Logger             hclog.Logger
+	ManagementClient   *management.Client
+}
+
+func (self *LlamaCppObserver) ObserveAndReport(
+	serverEventsChannel chan goroutine.ResultMessage,
+) {
+	llamaCppHealthStatusChannel := make(chan llamacpp.LlamaCppHealthStatus)
+
+	defer close(llamaCppHealthStatusChannel)
+
+	ticker := time.NewTicker(self.AgentConfiguration.GetReportingIntervalDuration())
+
+	go self.RunTickerInterval(llamaCppHealthStatusChannel, ticker)
+
+	for llamaCppHealthStatus := range llamaCppHealthStatusChannel {
+		go self.ManagementClient.ReportLlamaCppHealthStatus(
+			serverEventsChannel,
+			self.LlamaCppClient.LlamaCppConfiguration,
+			&llamaCppHealthStatus,
+		)
+	}
+}
+
+func (self *LlamaCppObserver) RunTickerInterval(
+	llamaCppHealthStatusChannel chan llamacpp.LlamaCppHealthStatus,
+	ticker *time.Ticker,
+) {
+	for range ticker.C {
+		go self.LlamaCppClient.GetHealth(llamaCppHealthStatusChannel)
+	}
 }
diff --git a/cmd/Agent.go b/cmd/Agent.go
@@ -1,18 +1,49 @@
 package cmd
 
 import (
+	"net/http"
+
+	"github.com/distantmagic/paddler/agent"
+	"github.com/distantmagic/paddler/goroutine"
 	"github.com/distantmagic/paddler/llamacpp"
+	"github.com/distantmagic/paddler/management"
 	"github.com/distantmagic/paddler/reverseproxy"
 	"github.com/hashicorp/go-hclog"
 	"github.com/urfave/cli/v2"
 )
 
 type Agent struct {
-	Logger                    hclog.Logger
-	LlamaCppConfiguration     *llamacpp.LlamaCppConfiguration
-	ReverseProxyConfiguration *reverseproxy.ReverseProxyConfiguration
+	AgentConfiguration            *agent.AgentConfiguration
+	Logger                        hclog.Logger
+	LlamaCppConfiguration         *llamacpp.LlamaCppConfiguration
+	ManagementServerConfiguration *management.ManagementServerConfiguration
+	ReverseProxyConfiguration     *reverseproxy.ReverseProxyConfiguration
 }
 
 func (self *Agent) Action(cliContext *cli.Context) error {
+	serverEventsChannel := make(chan goroutine.ResultMessage)
+
+	llamaCppObserver := &agent.LlamaCppObserver{
+		AgentConfiguration: self.AgentConfiguration,
+		LlamaCppClient: &llamacpp.LlamaCppClient{
+			HttpClient:            http.DefaultClient,
+			LlamaCppConfiguration: self.LlamaCppConfiguration,
+		},
+		Logger: self.Logger.Named("LlamaCppObserver"),
+		ManagementClient: &management.Client{
+			HttpClient:                    http.DefaultClient,
+			ManagementServerConfiguration: self.ManagementServerConfiguration,
+		},
+	}
+
+	go llamaCppObserver.ObserveAndReport(serverEventsChannel)
+
+	for serverEvent := range serverEventsChannel {
+		self.Logger.Info(
+			"server event",
+			"event", serverEvent,
+		)
+	}
+
 	return nil
 }
diff --git a/cmd/Balancer.go b/cmd/Balancer.go
@@ -37,7 +37,7 @@ func (self *Balancer) Action(cliContext *cli.Context) error {
 		},
 	}
 
-	reverseProxyServer := &reverseproxy.Server{
+	reverseProxyServer := &loadbalancer.ReverseProxyServer{
 		LoadBalancer:              loadBalancer,
 		Logger:                    self.Logger.Named("reverseproxy"),
 		ReverseProxyConfiguration: self.ReverseProxyConfiguration,
@@ -47,8 +47,9 @@ func (self *Balancer) Action(cliContext *cli.Context) error {
 	go reverseProxyServer.Serve(serverEventsChannel)
 
 	for serverEvent := range serverEventsChannel {
-		self.Logger.Info(
-			"server",
+		self.Logger.Log(
+			hclog.Info,
+			"server event",
 			"event", serverEvent,
 		)
 	}

diff --git a/llamacpp/LlamaCppClient.go b/llamacpp/LlamaCppClient.go
@@ -25,8 +25,6 @@ type LlamaCppClient struct {
 func (self *LlamaCppClient) GetHealth(
 	responseChannel chan LlamaCppHealthStatus,
 ) {
-	defer close(responseChannel)
-
 	request, err := http.NewRequest(
 		"GET",
 		self.LlamaCppConfiguration.HttpAddress.BuildUrlWithPath("health").String(),

diff --git a/llamacpp/LlamaCppClient_test.go b/llamacpp/LlamaCppClient_test.go
@@ -22,6 +22,8 @@ var llamaCppClient *LlamaCppClient = &LlamaCppClient{
 func TestHealthIsObtained(t *testing.T) {
 	responseChannel := make(chan LlamaCppHealthStatus)
 
+	defer close(responseChannel)
+
 	go llamaCppClient.GetHealth(responseChannel)
 
 	healthStatus := <-responseChannel

diff --git a/loadbalancer/LlamaCppTargetConfiguration.go b/loadbalancer/LlamaCppTargetConfiguration.go
@@ -3,5 +3,5 @@ package loadbalancer
 import "github.com/distantmagic/paddler/llamacpp"
 
 type LlamaCppTargetConfiguration struct {
-	LlamaCppConfiguration *llamacpp.LlamaCppConfiguration
+	LlamaCppConfiguration *llamacpp.LlamaCppConfiguration `json:"llama_cpp_configuration"`
 }
diff --git a/loadbalancer/LoadBalancerStatus.go b/loadbalancer/LoadBalancerStatus.go
@@ -1,5 +1,5 @@
 package loadbalancer
 
 type LoadBalancerStatus struct {
-	RegisteredTargets int
+	RegisteredTargets int `json:"registred_targets"`
 }
diff --git a/loadbalancer/ReverseProxyServer.go b/loadbalancer/ReverseProxyServer.go
@@ -0,0 +1,56 @@
+package loadbalancer
+
+import (
+	"net/http"
+	"net/http/httputil"
+
+	"github.com/distantmagic/paddler/goroutine"
+	"github.com/distantmagic/paddler/reverseproxy"
+	"github.com/hashicorp/go-hclog"
+)
+
+type ReverseProxyServer struct {
+	LoadBalancer              *LoadBalancer
+	Logger                    hclog.Logger
+	ReverseProxyConfiguration *reverseproxy.ReverseProxyConfiguration
+}
+
+func (self *ReverseProxyServer) Serve(serverEventsChannel chan goroutine.ResultMessage) {
+	self.Logger.Debug(
+		"listen",
+		"host", self.ReverseProxyConfiguration.HttpAddress.GetHostWithPort(),
+	)
+
+	reverseProxy := &httputil.ReverseProxy{
+		ErrorLog: self.Logger.Named("ReverseProxy").StandardLogger(&hclog.StandardLoggerOptions{
+			InferLevels: true,
+		}),
+		Rewrite: func(request *httputil.ProxyRequest) {
+			targetUrl, err := self.LoadBalancer.Balance(request.In)
+
+			if err != nil {
+				serverEventsChannel <- goroutine.ResultMessage{
+					Comment: "failed to balance request",
+					Error:   err,
+				}
+
+				return
+			}
+
+			request.SetURL(targetUrl)
+			request.SetXForwarded()
+		},
+	}
+
+	err := http.ListenAndServe(
+		self.ReverseProxyConfiguration.HttpAddress.GetHostWithPort(),
+		reverseProxy,
+	)
+
+	if err != nil {
+		serverEventsChannel <- goroutine.ResultMessage{
+			Comment: "failed to listen",
+			Error:   err,
+		}
+	}
+}
diff --git a/main.go b/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"os"
 
+	"github.com/distantmagic/paddler/agent"
 	"github.com/distantmagic/paddler/cmd"
 	"github.com/distantmagic/paddler/llamacpp"
 	"github.com/distantmagic/paddler/management"
@@ -12,17 +13,28 @@ import (
 	"github.com/urfave/cli/v2"
 )
 
+const (
+	DefaultManagementHost   = "127.0.0.1"
+	DefaultManagementPort   = 8085
+	DefaultManagementScheme = "http"
+)
+
 func main() {
 	logger := hclog.New(&hclog.LoggerOptions{
-		Name:  "paddler",
+		Name: "paddler",
+		// JSONFormat: true,
 		Level: hclog.Debug,
 	})
 
 	agent := &cmd.Agent{
-		Logger: logger.Named("Agent"),
+		AgentConfiguration: &agent.AgentConfiguration{},
 		LlamaCppConfiguration: &llamacpp.LlamaCppConfiguration{
 			HttpAddress: &netcfg.HttpAddressConfiguration{},
 		},
+		Logger: logger.Named("Agent"),
+		ManagementServerConfiguration: &management.ManagementServerConfiguration{
+			HttpAddress: &netcfg.HttpAddressConfiguration{},
+		},
 		ReverseProxyConfiguration: &reverseproxy.ReverseProxyConfiguration{
 			HttpAddress: &netcfg.HttpAddressConfiguration{},
 		},
@@ -47,20 +59,25 @@ func main() {
 				Usage:  "start llama.cpp observer agent",
 				Action: agent.Action,
 				Flags: []cli.Flag{
+					&cli.UintFlag{
+						Name:        "agent-reporting-interval-miliseconds",
+						Value:       1000,
+						Destination: &agent.AgentConfiguration.ReportingIntervalMiliseconds,
+					},
 					&cli.StringFlag{
-						Name:        "balancer-host",
-						Value:       "127.0.0.1",
-						Destination: &agent.ReverseProxyConfiguration.HttpAddress.Host,
+						Name:        "management-host",
+						Value:       DefaultManagementHost,
+						Destination: &agent.ManagementServerConfiguration.HttpAddress.Host,
 					},
 					&cli.UintFlag{
-						Name:        "balancer-port",
-						Value:       8085,
-						Destination: &agent.ReverseProxyConfiguration.HttpAddress.Port,
+						Name:        "management-port",
+						Value:       DefaultManagementPort,
+						Destination: &agent.ManagementServerConfiguration.HttpAddress.Port,
 					},
 					&cli.StringFlag{
-						Name:        "balancer-scheme",
-						Value:       "http",
-						Destination: &agent.ReverseProxyConfiguration.HttpAddress.Scheme,
+						Name:        "management-scheme",
+						Value:       DefaultManagementScheme,
+						Destination: &agent.ManagementServerConfiguration.HttpAddress.Scheme,
 					},
 					&cli.StringFlag{
 						Name:        "llamacpp-host",
@@ -69,14 +86,29 @@ func main() {
 					},
 					&cli.UintFlag{
 						Name:        "llamacpp-port",
-						Value:       8080,
+						Value:       8088,
 						Destination: &agent.LlamaCppConfiguration.HttpAddress.Port,
 					},
 					&cli.StringFlag{
 						Name:        "llamacpp-scheme",
 						Value:       "http",
 						Destination: &agent.LlamaCppConfiguration.HttpAddress.Scheme,
 					},
+					&cli.StringFlag{
+						Name:        "reverseproxy-host",
+						Value:       "127.0.0.1",
+						Destination: &balancer.ReverseProxyConfiguration.HttpAddress.Host,
+					},
+					&cli.UintFlag{
+						Name:        "reverseproxy-port",
+						Value:       8086,
+						Destination: &balancer.ReverseProxyConfiguration.HttpAddress.Port,
+					},
+					&cli.StringFlag{
+						Name:        "reverseproxy-scheme",
+						Value:       "http",
+						Destination: &balancer.ReverseProxyConfiguration.HttpAddress.Scheme,
+					},
 				},
 			},
 			{
@@ -86,17 +118,17 @@ func main() {
 				Flags: []cli.Flag{
 					&cli.StringFlag{
 						Name:        "management-host",
-						Value:       "127.0.0.1",
+						Value:       DefaultManagementHost,
 						Destination: &balancer.ManagementServerConfiguration.HttpAddress.Host,
 					},
 					&cli.UintFlag{
 						Name:        "management-port",
-						Value:       8085,
+						Value:       DefaultManagementPort,
 						Destination: &balancer.ManagementServerConfiguration.HttpAddress.Port,
 					},
 					&cli.StringFlag{
 						Name:        "management-scheme",
-						Value:       "http",
+						Value:       DefaultManagementScheme,
 						Destination: &balancer.ManagementServerConfiguration.HttpAddress.Scheme,
 					},
 					&cli.StringFlag{