mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-11 16:09:07 +02:00
[e2e] Add live chat-through-proxy scenario (Pillar 3)
Stand up the full agent-network data path in containers and drive a real chat-completion through the gateway: - harness: a shared docker network (combined server reachable by alias), a proxy container built from the published reverse-proxy image (NB_PROXY_PRIVATE, NB_PROXY_ALLOW_INSECURE, NB_RELAY_TRANSPORT=ws to match the combined server's WS-multiplexed relay) with a generated self-signed wildcard cert, and a netbird client container that joins via a setup key. - the combined image, proxy image, and client image default to the published rc.2 releases (overridable via NB_E2E_*_IMAGE; a bare local tag is built from source instead). Geolocation download is disabled so the server starts without external fetches. - one shared domain is used for the management exposed address, the proxy domain, and the agent-network cluster; the proxy token is minted via the server CLI (global) to match the manual install. TestChatCompletionThroughProxy provisions provider+policy+group+setup key, runs proxy+client, drives an OpenAI chat-completion through the tunnel, and asserts a 200 plus the ingested access-log row. Requires OPENAI_TOKEN (skips otherwise). The provider must be created with enabled=true explicitly — the create default is false despite the API doc.
This commit is contained in:
@@ -0,0 +1,207 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/api/types/container"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
tcexec "github.com/testcontainers/testcontainers-go/exec"
|
||||
)
|
||||
|
||||
const (
|
||||
clientDockerfile = "e2e/harness/Dockerfile.client"
|
||||
// defaultClientImage is the published NetBird client release used by
|
||||
// default. Override with NB_E2E_CLIENT_IMAGE; a value without a "/" is built
|
||||
// locally from clientDockerfile.
|
||||
defaultClientImage = "netbirdio/netbird:0.74.0-rc.2"
|
||||
clientAlias = "client"
|
||||
curlImage = "curlimages/curl:latest"
|
||||
)
|
||||
|
||||
// Client is a running NetBird client container joined to the combined server.
|
||||
type Client struct {
|
||||
container testcontainers.Container
|
||||
}
|
||||
|
||||
// StartClient builds the client image and runs it on the combined server's
|
||||
// network, joining via the given setup key. The image entrypoint brings the
|
||||
// daemon up automatically; callers wait for connectivity with WaitConnected /
|
||||
// WaitProxyPeer.
|
||||
func StartClient(ctx context.Context, c *Combined, setupKey string) (*Client, error) {
|
||||
root, err := repoRoot()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clientImage, err := resolveImage(ctx, root, "NB_E2E_CLIENT_IMAGE", defaultClientImage, clientDockerfile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req := testcontainers.ContainerRequest{
|
||||
Image: clientImage,
|
||||
Networks: []string{c.network.Name},
|
||||
NetworkAliases: map[string][]string{c.network.Name: {clientAlias}},
|
||||
Env: map[string]string{
|
||||
"NB_MANAGEMENT_URL": combinedExposedURL,
|
||||
"NB_SETUP_KEY": setupKey,
|
||||
"NB_LOG_LEVEL": "info",
|
||||
// Match the proxy: the combined relay is WebSocket-only, so the
|
||||
// client must use WS transport to keep a stable relay link to it.
|
||||
"NB_RELAY_TRANSPORT": "ws",
|
||||
},
|
||||
HostConfigModifier: func(hc *container.HostConfig) {
|
||||
hc.CapAdd = append(hc.CapAdd, "NET_ADMIN", "SYS_ADMIN", "SYS_RESOURCE")
|
||||
},
|
||||
}
|
||||
|
||||
ctr, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
|
||||
ContainerRequest: req,
|
||||
Started: true,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("start client container: %w", err)
|
||||
}
|
||||
return &Client{container: ctr}, nil
|
||||
}
|
||||
|
||||
// Restart bounces the client connection (netbird down/up) so it pulls a fresh
|
||||
// network map — the documented workaround for a freshly-joined client not yet
|
||||
// seeing a synthesized agent-network service.
|
||||
func (cl *Client) Restart(ctx context.Context) error {
|
||||
if _, _, err := cl.container.Exec(ctx, []string{"netbird", "down"}, tcexec.Multiplexed()); err != nil {
|
||||
return fmt.Errorf("netbird down: %w", err)
|
||||
}
|
||||
time.Sleep(2 * time.Second)
|
||||
code, reader, err := cl.container.Exec(ctx, []string{"netbird", "up"}, tcexec.Multiplexed())
|
||||
if err != nil {
|
||||
return fmt.Errorf("netbird up: %w", err)
|
||||
}
|
||||
if code != 0 {
|
||||
out, _ := io.ReadAll(reader)
|
||||
return fmt.Errorf("netbird up exited %d: %s", code, string(out))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Status returns `netbird status` output from inside the client.
|
||||
func (cl *Client) Status(ctx context.Context) (string, error) {
|
||||
code, reader, err := cl.container.Exec(ctx, []string{"netbird", "status"}, tcexec.Multiplexed())
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
out, _ := io.ReadAll(reader)
|
||||
if code != 0 {
|
||||
return string(out), fmt.Errorf("netbird status exited %d", code)
|
||||
}
|
||||
return string(out), nil
|
||||
}
|
||||
|
||||
// WaitConnected polls until the client reports Management: Connected.
|
||||
func (cl *Client) WaitConnected(ctx context.Context, timeout time.Duration) error {
|
||||
return cl.pollStatus(ctx, timeout, "Management: Connected")
|
||||
}
|
||||
|
||||
// WaitProxyPeer polls until the client sees the proxy peer connected (1/1).
|
||||
func (cl *Client) WaitProxyPeer(ctx context.Context, timeout time.Duration) error {
|
||||
return cl.pollStatus(ctx, timeout, "1/1 Connected")
|
||||
}
|
||||
|
||||
func (cl *Client) pollStatus(ctx context.Context, timeout time.Duration, want string) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
var last string
|
||||
for time.Now().Before(deadline) {
|
||||
out, _ := cl.Status(ctx)
|
||||
last = out
|
||||
if strings.Contains(out, want) {
|
||||
return nil
|
||||
}
|
||||
time.Sleep(3 * time.Second)
|
||||
}
|
||||
return fmt.Errorf("timed out waiting for %q; last status:\n%s", want, last)
|
||||
}
|
||||
|
||||
// ResolveProxyIP resolves the agent-network endpoint to the proxy peer's
|
||||
// NetBird IP from inside the client (via magic DNS).
|
||||
func (cl *Client) ResolveProxyIP(ctx context.Context, endpoint string) (string, error) {
|
||||
code, reader, err := cl.container.Exec(ctx, []string{"getent", "hosts", endpoint}, tcexec.Multiplexed())
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
out, _ := io.ReadAll(reader)
|
||||
if code != 0 {
|
||||
return "", fmt.Errorf("getent hosts %s exited %d", endpoint, code)
|
||||
}
|
||||
fields := strings.Fields(string(out))
|
||||
if len(fields) == 0 {
|
||||
return "", fmt.Errorf("no address for %s", endpoint)
|
||||
}
|
||||
return fields[0], nil
|
||||
}
|
||||
|
||||
// Chat issues a chat-completion POST to the agent-network endpoint over the
|
||||
// client's tunnel, returning the HTTP status and response body. It runs curl in
|
||||
// a throwaway container sharing the client's network namespace so the request
|
||||
// traverses the WireGuard tunnel, pinning the endpoint to the proxy peer IP.
|
||||
func (cl *Client) Chat(ctx context.Context, endpoint, proxyIP, model, prompt string) (int, string, error) {
|
||||
body := fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":%q}]}`, model, prompt)
|
||||
url := "https://" + endpoint + "/v1/chat/completions"
|
||||
|
||||
args := []string{
|
||||
"run", "--rm",
|
||||
"--network", "container:" + cl.container.GetContainerID(),
|
||||
curlImage,
|
||||
"-sk", "--connect-timeout", "5", "--max-time", "90",
|
||||
"--resolve", endpoint + ":443:" + proxyIP,
|
||||
"-o", "/dev/stderr", "-w", "%{http_code}",
|
||||
"-X", "POST", url,
|
||||
"-H", "Content-Type: application/json",
|
||||
"--data", body,
|
||||
}
|
||||
cmd := exec.CommandContext(ctx, "docker", args...)
|
||||
// -w writes the status code to stdout; -o /dev/stderr writes the body to
|
||||
// stderr so we can capture both separately.
|
||||
var stdout, stderr strings.Builder
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
if err := cmd.Run(); err != nil {
|
||||
return 0, stderr.String(), fmt.Errorf("curl through tunnel: %w", err)
|
||||
}
|
||||
|
||||
code := 0
|
||||
_, _ = fmt.Sscanf(strings.TrimSpace(stdout.String()), "%d", &code)
|
||||
return code, stderr.String(), nil
|
||||
}
|
||||
|
||||
// Logs returns the client container logs, for diagnostics on failure.
|
||||
func (cl *Client) Logs(ctx context.Context) string {
|
||||
return containerLogs(ctx, cl.container)
|
||||
}
|
||||
|
||||
// Terminate stops the client container.
|
||||
func (cl *Client) Terminate(ctx context.Context) error {
|
||||
if cl.container == nil {
|
||||
return nil
|
||||
}
|
||||
return cl.container.Terminate(ctx)
|
||||
}
|
||||
|
||||
// containerLogs reads up to 256 KiB of a container's logs for diagnostics.
|
||||
func containerLogs(ctx context.Context, c testcontainers.Container) string {
|
||||
if c == nil {
|
||||
return ""
|
||||
}
|
||||
r, err := c.Logs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("<logs error: %v>", err)
|
||||
}
|
||||
defer r.Close()
|
||||
b, _ := io.ReadAll(io.LimitReader(r, 256<<10))
|
||||
return string(b)
|
||||
}
|
||||
Reference in New Issue
Block a user