Websocket logs endpoint for polling envoy fleet container logs

.github/workflows/build.yml

@@ -24,15 +24,19 @@ jobs:
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v1
 
-      - name: Build and push
+      - name: Build and push api-server
         uses: docker/build-push-action@v2
         with:
-          context: server
-          file: server/Dockerfile
+          context: .
+          file: ./build/api-server/Dockerfile
           platforms: linux/amd64,linux/arm64
 
-#      - name: build
-#        run: make build
+      - name: Build and push websocket
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          file: ./build/websocket/Dockerfile
+          platforms: linux/amd64,linux/arm64
 
       - name: Set up Go
         uses: actions/setup-go@v2

.github/workflows/release.yml

@@ -51,11 +51,23 @@ jobs:
       - name: setup-goreleaser-environment-variables
         run: |
           echo "VERSION=$(git describe --tags $(git rev-list --tags --max-count=1))" >> $GITHUB_ENV
-      - name: Build and Push
+      - name: Build and Push Kusk Gateway API
         uses: docker/build-push-action@v2
         with:
-          context: server
-          file: server/Dockerfile
+          context: .
+          file: ./build/api-server/Dockerfile
+          push: true
+          platforms: linux/amd64,linux/arm64
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          build-args: |
+            TELEMETRY_TOKEN=${{ secrets.TELEMETRY_TOKEN }}
+            VERSION=${{ env.VERSION }}
+      - name: Build and Push Kusk Gateway API Websocket
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          file: ./build/websocket/Dockerfile
           push: true
           platforms: linux/amd64,linux/arm64
           tags: ${{ steps.meta.outputs.tags }}

Makefile

@@ -1,8 +1,10 @@
 .PHONY: all
 all: format test build
 
+.PHONY: build
 build:
-	docker build -t kusk-gateway-api server
+	docker buildx build -t kubeshop/kusk-gateway-api -f build/api-server/Dockerfile .
+	docker buildx build -t kubeshop/kusk-gateway-api-websocket -f build/websocket/Dockerfile .
 
 server-generate:
 	openapi-generator-cli generate -i api/openapi.yaml -g go-server -o server/ --additional-properties=featureCORS=true

api/openapi.yaml

@@ -425,6 +425,45 @@ paths:
                 type: object
         404:
           description: "Envoy fleet CRD not found"
+  /logs:
+    x-kusk:
+      upstream:
+        service:
+          name: kusk-gateway-api
+          namespace: kusk-system
+          port: 8081
+        rewrite:
+          pattern: "^/api"
+          substitution: ""
+      websocket: true
+    get:
+      tags:
+        - "fleets"
+      summary: "Get envoy fleet logs"
+      operationId: getEnvoyFleetLogs
+      parameters:
+        - name: namespace
+          in: query
+          required: true
+          schema:
+            type: string
+            default: "kusk-system"
+        - name: name
+          in: query
+          required: true
+          schema:
+            type: string
+            default: "kusk-gateway-envoy-fleet"
+      responses:
+        200:
+          description: "Envoy fleet logs"
+          content:
+            application/json:
+              schema:
+                type: object
+        404:
+          description: "Envoy fleet logs not found"
+
   /staticroutes:
     get:
       tags:

server/Dockerfile → build/api-server/Dockerfile

@@ -2,7 +2,7 @@ FROM --platform=$BUILDPLATFORM docker.io/golang:1.18 as builder
 WORKDIR /go/src
 # Copy `go.mod` for definitions and `go.sum` to invalidate the next layer
 # in case of a change in the dependencies
-COPY go.mod go.sum ./
+COPY ./server/go.mod ./server/go.sum ./
 # Download dependencies
 RUN go mod download
 
@@ -11,8 +11,8 @@ ARG VERSION
 ARG TARGETARCH
 ARG TARGETOS
 
-COPY . .
-RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -v -ldflags "-X github.com/kubeshop/kusk-gateway/pkg/analytics.TelemetryToken=$TELEMETRY_TOKEN -X github.com/kubeshop/kusk-gateway/pkg/build.Version=$VERSION" -o kusk-gateway-api
+COPY ./server ./
+RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -v -ldflags "-X github.com/kubeshop/kusk-gateway/pkg/analytics.TelemetryToken=$TELEMETRY_TOKEN -X github.com/kubeshop/kusk-gateway/pkg/build.Version=$VERSION" -o kusk-gateway-api cmd/api-server/main.go
 
 FROM --platform=$BUILDPLATFORM gcr.io/distroless/static:nonroot
 COPY --from=builder --chown=65532:65532 /go/src/kusk-gateway-api ./

build/websocket/Dockerfile

@@ -0,0 +1,22 @@
+FROM --platform=$BUILDPLATFORM docker.io/golang:1.18 as builder
+WORKDIR /go/src
+# Copy `go.mod` for definitions and `go.sum` to invalidate the next layer
+# in case of a change in the dependencies
+COPY ../server/go.mod ../server/go.sum ./
+# Download dependencies
+RUN go mod download
+
+ARG TELEMETRY_TOKEN
+ARG VERSION
+ARG TARGETARCH
+ARG TARGETOS
+
+COPY ../server ./
+RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -v -ldflags "-X github.com/kubeshop/kusk-gateway/pkg/analytics.TelemetryToken=$TELEMETRY_TOKEN -X github.com/kubeshop/kusk-gateway/pkg/build.Version=$VERSION" -o kusk-gateway-api-websocket cmd/websocket/*.go
+
+FROM --platform=$BUILDPLATFORM gcr.io/distroless/static:nonroot
+COPY --from=builder --chown=65532:65532 /go/src/kusk-gateway-api-websocket ./
+
+USER 65532:65532
+
+ENTRYPOINT ["./kusk-gateway-api-websocket"]

docker-compose-minikube.yaml

@@ -3,3 +3,6 @@ services:
   kgwapi:
     volumes:
       - $HOME/.minikube:$HOME/.minikube:ro
+  websocket:
+    volumes:
+      - $HOME/.minikube:$HOME/.minikube:ro

docker-compose.yaml

@@ -1,11 +1,23 @@
 version: "3"
 services:
   kgwapi:
-    build: ./server
+    build:
+      context: .
+      dockerfile: ./build/api-server/Dockerfile
     environment:
       - KUBECONFIG=/kube/config
       - ANALYTICS_ENABLED=false
     volumes:
       - $HOME/.kube/config:/kube/config:ro
     ports:
-      - 8080:8080
+      - "8080:8080"
+  websocket:
+    build:
+      context: .
+      dockerfile: ./build/websocket/Dockerfile
+    environment:
+      - KUBECONFIG=/kube/config
+    volumes:
+      - $HOME/.kube/config:/kube/config:ro
+    ports:
+        - "8081:8080"

kgw-api.yaml

@@ -17,9 +17,15 @@ spec:
         app: kgwtest
     spec:
       containers:
-        - image: jasmingacic/kusk-gateway-api:latest
-          imagePullPolicy: Always
-          name: kuskgateway-api
+        - image: kubeshop/kusk-gateway-api:latest
+          imagePullPolicy: IfNotPresent
+          name: kusk-gateway-api
+        - image: kubeshop/kusk-gateway-api-websocket:latest
+          imagePullPolicy: IfNotPresent
+          args:
+            - --port=8081
+          name: kusk-gateway-api-websocket
+
       serviceAccountName: kusk-gateway-manager
 ---
 apiVersion: v1
@@ -32,7 +38,11 @@ metadata:
 spec:
   ports:
     - port: 8080
+      name: http
       targetPort: 8080
+    - port: 8081
+      name: websocket
+      targetPort: 8081
   type: LoadBalancer
   selector:
     app: kgwtest

server/main.go → server/cmd/api-server/main.go

@@ -137,14 +137,20 @@ func getConfig() (*rest.Config, error) {
 
 	return config, err
 }
+
 func getClient() (client.Client, error) {
 	scheme := runtime.NewScheme()
-	kuskv1.AddToScheme(scheme)
-	corev1.AddToScheme(scheme)
+	if err := kuskv1.AddToScheme(scheme); err != nil {
+		return nil, fmt.Errorf("unable to add kusk scheme: %w", err)
+	}
+
+	if err := corev1.AddToScheme(scheme); err != nil {
+		return nil, fmt.Errorf("unable to add core scheme: %w", err)
+	}
 
 	config, err := getConfig()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("unable to get kubernetes config: %w", err)
 	}
 
 	return client.New(config, client.Options{Scheme: scheme})

server/cmd/websocket/client.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"log"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+type client struct {
+	conn      *websocket.Conn
+	logStream io.ReadCloser
+
+	writeWait      time.Duration
+	pongWait       time.Duration
+	pingPeriod     time.Duration
+	maxMessageSize int64
+}
+
+func (c *client) readPump(ctx context.Context, stopCh chan struct{}) {
+	defer func() {
+		c.conn.Close()
+		stopCh <- struct{}{}
+	}()
+	c.conn.SetReadLimit(c.maxMessageSize)
+	c.conn.SetReadDeadline(time.Now().Add(c.pongWait))
+	c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(c.pongWait)); return nil })
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_, _, err := c.conn.ReadMessage()
+			if err != nil {
+				if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, websocket.CloseNormalClosure) {
+					log.Printf("error: %v", err)
+				}
+				return
+			}
+		}
+	}
+}
+
+func (c *client) writePump(ctx context.Context, stopCh chan struct{}) {
+	ticker := time.NewTicker(c.pingPeriod)
+	defer func() {
+		ticker.Stop()
+		c.conn.Close()
+		stopCh <- struct{}{}
+	}()
+
+	reader := bufio.NewReader(c.logStream)
+	for {
+		select {
+		case <-ctx.Done():
+		case <-stopCh:
+			return
+		case <-ticker.C:
+			c.conn.SetWriteDeadline(time.Now().Add(c.writeWait))
+			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		default:
+			c.conn.SetWriteDeadline(time.Now().Add(c.writeWait))
+			line, err := readLongLine(reader)
+			if err != nil {
+				log.Println("writePump: cannot read line", err)
+				continue
+			}
+
+			if err := c.conn.WriteMessage(websocket.TextMessage, line); err != nil {
+				log.Println("writePump: cannot write message", err)
+				continue
+			}
+		}
+	}
+}
+
+func readLongLine(r *bufio.Reader) (line []byte, err error) {
+	var buffer []byte
+	var isPrefix bool
+
+	for {
+		buffer, isPrefix, err = r.ReadLine()
+		line = append(line, buffer...)
+		if err != nil {
+			break
+		}
+
+		if !isPrefix {
+			break
+		}
+	}
+
+	return line, err
+}