support regional log pulls for cloud agents (#695)

* support regional log pulls for cloud agents

* update proto version

* fix log call when agent doesn't exist

* update var name

Author: real-danm

cmd/lk/agent.go

@@ -572,7 +572,7 @@ func createAgent(ctx context.Context, cmd *cli.Command) error {
 			return err
 		} else if viewLogs {
 			fmt.Println("Tailing runtime logs...safe to exit at any time")
-			return agentsClient.StreamLogs(ctx, "deploy", lkConfig.Agent.ID, os.Stdout)
+			return agentsClient.StreamLogs(ctx, "deploy", lkConfig.Agent.ID, os.Stdout, resp.ServerRegions[0])
 		}
 	}
 	return nil
@@ -859,7 +859,19 @@ func getLogs(ctx context.Context, cmd *cli.Command) error {
 	if err != nil {
 		return err
 	}
-	return agentsClient.StreamLogs(ctx, cmd.String("log-type"), agentID, os.Stdout)
+
+	response, err := agentsClient.ListAgents(ctx, &lkproto.ListAgentsRequest{
+		AgentId: agentID,
+	})
+	if err != nil {
+		return err
+	}
+
+	if len(response.Agents) == 0 {
+		return fmt.Errorf("no agent deployments found")
+	}
+
+	return agentsClient.StreamLogs(ctx, cmd.String("log-type"), agentID, os.Stdout, response.Agents[0].AgentDeployments[0].ServerRegion)
 }
 
 func deleteAgent(ctx context.Context, cmd *cli.Command) error {

pkg/agentfs/client.go

@@ -23,12 +23,13 @@ import (
 	"regexp"
 	"strings"
 
+	"github.com/twitchtv/twirp"
+
 	livekitcli "github.com/livekit/livekit-cli/v2"
 	"github.com/livekit/protocol/auth"
 	lkproto "github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	lksdk "github.com/livekit/server-sdk-go/v2"
-	"github.com/twitchtv/twirp"
 )
 
 // Client is a wrapper around the lksdk.AgentClient that provides a simpler interface for creating and deploying agents.
@@ -63,7 +64,7 @@ func New(opts ...ClientOption) (*Client, error) {
 		return nil, err
 	}
 	client.AgentClient = agentClient
-	client.agentsURL = client.getAgentsURL()
+	client.agentsURL = client.getAgentsURL("")
 	if client.httpClient == nil {
 		client.httpClient = &http.Client{}
 	}
@@ -172,7 +173,7 @@ func (c *Client) uploadAndBuild(
 	return nil
 }
 
-func (c *Client) getAgentsURL() string {
+func (c *Client) getAgentsURL(serverRegion string) string {
 	agentsURL := c.projectURL
 	if strings.HasPrefix(agentsURL, "ws") {
 		agentsURL = strings.Replace(agentsURL, "ws", "http", 1)
@@ -182,7 +183,10 @@ func (c *Client) getAgentsURL() string {
 	} else if !strings.Contains(agentsURL, "localhost") && !strings.Contains(agentsURL, "127.0.0.1") {
 		pattern := `^https://[a-zA-Z0-9\-]+\.`
 		re := regexp.MustCompile(pattern)
-		agentsURL = re.ReplaceAllString(agentsURL, "https://agents.")
+		if serverRegion != "" {
+			serverRegion = fmt.Sprintf("%s.", serverRegion)
+		}
+		agentsURL = re.ReplaceAllString(agentsURL, fmt.Sprintf("https://%sagents.", serverRegion))
 	}
 	return agentsURL
 }

pkg/agentfs/logs.go

@@ -33,15 +33,15 @@ type APIError struct {
 }
 
 // StreamLogs streams the logs for the given agent.
-func (c *Client) StreamLogs(ctx context.Context, logType, agentID string, writer io.Writer) error {
+func (c *Client) StreamLogs(ctx context.Context, logType, agentID string, writer io.Writer, serverRegion string) error {
 	logger := c.logger.WithName("StreamLogs")
 	if logType == "" {
 		logType = "deploy"
 	}
 	params := url.Values{}
 	params.Add("agent_id", agentID)
 	params.Add("log_type", logType)
-	fullUrl := fmt.Sprintf("%s/logs?%s", c.agentsURL, params.Encode())
+	fullUrl := fmt.Sprintf("%s/logs?%s", c.getAgentsURL(serverRegion), params.Encode())
 	req, err := c.newRequest("GET", fullUrl, nil)
 	if err != nil {
 		return err