Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Admin Bot
|
cc9a32f53a | ||
 
|
7de71180b3 |
@@ -0,0 +1,4 @@
|
||||
(apply,CacheStats{hitCount=337, missCount=199, loadSuccessCount=199, loadExceptionCount=0, totalLoadTime=581291927, evictionCount=0})
|
||||
(tree,CacheStats{hitCount=986, missCount=352, loadSuccessCount=299, loadExceptionCount=0, totalLoadTime=821650758, evictionCount=0})
|
||||
(commit,CacheStats{hitCount=108, missCount=107, loadSuccessCount=107, loadExceptionCount=0, totalLoadTime=78983052, evictionCount=0})
|
||||
(tag,CacheStats{hitCount=0, missCount=2, loadSuccessCount=2, loadExceptionCount=0, totalLoadTime=319542, evictionCount=0})
|
||||
@@ -0,0 +1,4 @@
|
||||
e71e5b78236a67327c678490cb50b46981f19de0 bbcbb68b91e786eb71bbb0a4443d7b8a26140e1b .sops.yaml
|
||||
4189696f5581ac0ffdc125c3bf9b9f664b3ddfb0 7cd3f1ee4865c563d141464f6fc185436993b84b .sops.yaml
|
||||
635630e73152a5f22e6cbd42322ec55d79f8d9c0 297e94a89d73d18c4f47013bb0e8303f123715f3 configmap.yaml
|
||||
29e515e7b46742fab8c3fcc2189af7010a6ccc62 6869fa11f96e03f7ec76a0ea14a4ddaf604004a4 gateway-config-secret.enc.yaml
|
||||
@@ -0,0 +1,12 @@
|
||||
0a95af80c0051bacbeb8483c1632e47acd3db5be 40207e487cfb63409a976fb2a0b9e1e62c8b1513
|
||||
27428d910111299d0699f429190284a9ca6e50b7 3318daf758349402aef43b095482743ab96b37f9
|
||||
329a495af4c935529fdae17229314101c0c77876 67f24ea76359c8dba4b56267790aad76bbc58464
|
||||
4c8bc6c920b6b75399555827022f69ef0c4f7d15 1fa839b41975fa3f0ac9052355ffb625f5a8f324
|
||||
528545f414c83217408edfea234dcd1f3edee0c2 b8f95506ca1545b876b5531cd385172e9ca5b4b0
|
||||
81038e1cf7567a9133d7c233a97b1e2f19fa1c82 4a00312906ba725f3968187656fde2663b1763ab
|
||||
a5b3b5c44a406896bcb414df6c6426c277715706 2ab47a9dbe5ba36dfa0e275991ef7b7656908410
|
||||
ce27643667a0399115cd1f2b6d38123fdcf2b4f1 6ff0a50de8efbad105fa588245f22fdb26afddc4
|
||||
d49756886a46542b38533b913a1f776b5145f5ec d82cc5a6970a1fb32e21dda9a737b987a8668111
|
||||
db3a30fbcf1f139c667fb68a91762582c49b8cee 04619a269fed9eeea53ab4d4d73131e3713f40a0
|
||||
eb54715e4dec0fb35402576fcc224a09808b00c1 d53b7632cf9646dda1c978a5f94615dc9eaed5e8
|
||||
ef72b5bbccf2df89aa1c86dee29311c63f33bf62 ba55d184fefef1a73a50409ca4fb1f7b27f5b075
|
||||
+71
-64
@@ -17,10 +17,13 @@ jobs:
|
||||
name: CI
|
||||
runs-on: golang
|
||||
steps:
|
||||
- name: Install Node.js and Docker
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
apt-get update
|
||||
apt-get install -y nodejs docker.io
|
||||
apt-get install -y docker.io curl nodejs
|
||||
curl -sLO "https://dl.k8s.io/release/$(curl -sL https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
|
||||
chmod +x kubectl && mv kubectl /usr/local/bin/
|
||||
kubectl version --client
|
||||
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
@@ -48,79 +51,83 @@ jobs:
|
||||
docker build --no-cache \
|
||||
-t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
|
||||
-f Dockerfile .
|
||||
echo "Built image: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
|
||||
- name: Push test image (SHA tag only, not latest yet)
|
||||
run: |
|
||||
docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
echo "✓ Pushed test image: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
- name: Push image (SHA tag)
|
||||
run: docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
|
||||
- name: Detect in-cluster Kubernetes authentication
|
||||
# ── Tekton integration tests ─────────────────────────────
|
||||
- name: Setup kubeconfig
|
||||
run: |
|
||||
# When running inside K8s cluster, kubectl auto-detects service account
|
||||
# Mounted at: /var/run/secrets/kubernetes.io/serviceaccount/
|
||||
if [ -f /var/run/secrets/kubernetes.io/serviceaccount/token ]; then
|
||||
echo "✓ In-cluster authentication detected"
|
||||
export KUBECONFIG=/dev/null # kubectl will auto-use in-cluster auth
|
||||
mkdir -p ~/.kube
|
||||
echo "${KUBECONFIG_B64}" | base64 -d > ~/.kube/config
|
||||
kubectl get pipelineruns -n api --no-headers | head -1 || echo 'No PipelineRuns yet'
|
||||
echo '✓ kubeconfig works'
|
||||
env:
|
||||
KUBECONFIG_B64: ${{ secrets.KUBECONFIG_B64 }}
|
||||
|
||||
- name: Trigger Tekton PipelineRun
|
||||
id: tekton
|
||||
run: |
|
||||
SHA="${{ steps.sha.outputs.short_sha }}"
|
||||
RUN_NAME="integration-test-${SHA}"
|
||||
|
||||
# Clean up any previous run with the same name
|
||||
kubectl delete taskrun "${RUN_NAME}" -n api --ignore-not-found
|
||||
|
||||
# Create TaskRun — spins up gateway sidecar + curl tests
|
||||
cat <<YAML | kubectl create -f -
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: TaskRun
|
||||
metadata:
|
||||
name: ${RUN_NAME}
|
||||
namespace: api
|
||||
labels:
|
||||
commit-sha: "${SHA}"
|
||||
spec:
|
||||
taskRef:
|
||||
name: integration-test
|
||||
params:
|
||||
- name: image
|
||||
value: "${IMAGE}:${SHA}"
|
||||
YAML
|
||||
|
||||
echo "✓ TaskRun created: ${RUN_NAME}"
|
||||
|
||||
# Wait for completion (Succeeded or Failed)
|
||||
echo "Waiting for tests (timeout 5m)..."
|
||||
if kubectl wait taskrun/"${RUN_NAME}" -n api \
|
||||
--for=condition=Succeeded --timeout=5m 2>/dev/null; then
|
||||
echo "result=pass" >> $GITHUB_OUTPUT
|
||||
else
|
||||
echo "⚠ Not running in-cluster, kubectl may fail"
|
||||
echo "result=fail" >> $GITHUB_OUTPUT
|
||||
fi
|
||||
|
||||
- name: Run integration tests via Kubernetes Job
|
||||
# Print logs + results
|
||||
echo ""
|
||||
echo "=== Test Logs ==="
|
||||
POD=$(kubectl get pod -n api -l tekton.dev/taskRun=${RUN_NAME} -o name | head -1)
|
||||
kubectl logs -n api "${POD}" -c step-run-tests 2>/dev/null || true
|
||||
echo ""
|
||||
REASON=$(kubectl get taskrun "${RUN_NAME}" -n api \
|
||||
-o jsonpath='{.status.conditions[0].reason}')
|
||||
SUMMARY=$(kubectl get taskrun "${RUN_NAME}" -n api \
|
||||
-o jsonpath='{.status.results[?(@.name=="summary")].value}')
|
||||
echo "Status: ${REASON}"
|
||||
echo "Summary: ${SUMMARY}"
|
||||
|
||||
- name: Gate on test result
|
||||
if: steps.tekton.outputs.result != 'pass'
|
||||
run: |
|
||||
echo "Running integration tests via Kubernetes Job..."
|
||||
echo "Test image: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
|
||||
# Apply job template from repo (uses in-cluster auth automatically)
|
||||
kubectl apply -f k8s/integration-test-job.yaml
|
||||
|
||||
# Update job to use new image
|
||||
kubectl set image job/api-gateway-integration-test \
|
||||
integration-tester="${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
|
||||
-n api --record
|
||||
|
||||
# Wait for job to complete (max 10 minutes)
|
||||
echo "Waiting for job to complete (this may take a few minutes)..."
|
||||
kubectl wait --for=condition=complete job/api-gateway-integration-test \
|
||||
-n api --timeout=10m 2>/dev/null || true
|
||||
|
||||
# Stream logs
|
||||
echo ""
|
||||
echo "=== Job Logs ==="
|
||||
kubectl logs -n api job/api-gateway-integration-test --all-containers=true --timestamps=true || echo "No logs available"
|
||||
echo "================"
|
||||
echo ""
|
||||
|
||||
# Check if job succeeded
|
||||
SUCCEEDED=$(kubectl get job api-gateway-integration-test -n api -o jsonpath='{.status.succeeded}' 2>/dev/null || echo "0")
|
||||
FAILED=$(kubectl get job api-gateway-integration-test -n api -o jsonpath='{.status.failed}' 2>/dev/null || echo "0")
|
||||
|
||||
echo "Job Status: Succeeded=$SUCCEEDED, Failed=$FAILED"
|
||||
|
||||
if [ "$SUCCEEDED" = "1" ]; then
|
||||
echo "✓ Integration tests PASSED"
|
||||
exit 0
|
||||
else
|
||||
echo "✗ Integration tests FAILED"
|
||||
echo "✗ Integration tests FAILED — image NOT promoted"
|
||||
exit 1
|
||||
fi
|
||||
continue-on-error: false
|
||||
|
||||
- name: Promote image to latest (only if tests passed)
|
||||
if: success()
|
||||
# ── Promote only after tests pass ────────────────────────
|
||||
- name: Promote image to latest
|
||||
run: |
|
||||
docker pull "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
docker tag "${IMAGE}:${{ steps.sha.outputs.short_sha }}" "${IMAGE}:latest"
|
||||
docker push "${IMAGE}:latest"
|
||||
echo "✓ Promoted ${IMAGE}:${{ steps.sha.outputs.short_sha }} to latest"
|
||||
echo "✓ Promoted to latest"
|
||||
|
||||
- name: Cleanup integration test job
|
||||
- name: Cleanup
|
||||
if: always()
|
||||
run: |
|
||||
echo "Cleaning up test job..."
|
||||
kubectl delete job api-gateway-integration-test -n api --ignore-not-found=true
|
||||
continue-on-error: true
|
||||
|
||||
- name: Cleanup docker
|
||||
if: always()
|
||||
run: docker image prune -a --force 2>&1 | tail -3 || true
|
||||
run: docker image prune -af 2>&1 | tail -3 || true
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
#!/bin/bash
|
||||
# Example: Send email via notification/sendMsg endpoint
|
||||
|
||||
BASE_URL="${1:-https://api.riotpiao.com}"
|
||||
AUTH_TOKEN="${2:-}" # Optional JWT token if auth required
|
||||
|
||||
PAYLOAD=$(cat <<'EOF'
|
||||
{
|
||||
"format": "smtp",
|
||||
"title": "System Alert",
|
||||
"message": "CPU usage exceeded 90% threshold",
|
||||
"priority": 7,
|
||||
"extras": {
|
||||
"to_email": "[email protected]",
|
||||
"cc": "[email protected]"
|
||||
}
|
||||
}
|
||||
EOF
|
||||
)
|
||||
|
||||
if [ -n "$AUTH_TOKEN" ]; then
|
||||
curl -X POST "$BASE_URL" \
|
||||
-H "X-Service: notification" \
|
||||
-H "X-Resource: sendMsg" \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer $AUTH_TOKEN" \
|
||||
-d "$PAYLOAD"
|
||||
else
|
||||
curl -X POST "$BASE_URL" \
|
||||
-H "X-Service: notification" \
|
||||
-H "X-Resource: sendMsg" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "$PAYLOAD"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
@@ -0,0 +1,160 @@
|
||||
package notification
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/smtp"
|
||||
"os"
|
||||
)
|
||||
|
||||
// SendMsgRequest represents a sendMsg API request.
|
||||
type SendMsgRequest struct {
|
||||
Format string `json:"format"` // "smtp" or "sms"
|
||||
Title string `json:"title"`
|
||||
Message string `json:"message"`
|
||||
Priority int `json:"priority,omitempty"`
|
||||
Extras map[string]string `json:"extras,omitempty"` // e.g., {"to_email": "[email protected]", "phone": "+1234567890"}
|
||||
}
|
||||
|
||||
// SendMsgResponse represents a sendMsg API response.
|
||||
type SendMsgResponse struct {
|
||||
Status string `json:"status"`
|
||||
MessageID string `json:"messageId,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// Handler handles sendMsg requests and forwards to appropriate channel (email, SMS, or Gotify push).
|
||||
type Handler struct {
|
||||
smtpHost string
|
||||
smtpPort string
|
||||
smtpFrom string
|
||||
smtpUser string
|
||||
smtpPass string
|
||||
smsAPIURL string
|
||||
smsAPIKey string
|
||||
gotifyURL string
|
||||
gotifyToken string
|
||||
}
|
||||
|
||||
// NewHandler creates a new notification handler from environment variables.
|
||||
func NewHandler() *Handler {
|
||||
return &Handler{
|
||||
smtpHost: os.Getenv("SMTP_HOST"),
|
||||
smtpPort: os.Getenv("SMTP_PORT"),
|
||||
smtpFrom: os.Getenv("SMTP_FROM"),
|
||||
smtpUser: os.Getenv("SMTP_USER"),
|
||||
smtpPass: os.Getenv("SMTP_PASS"),
|
||||
smsAPIURL: os.Getenv("SMS_API_URL"),
|
||||
smsAPIKey: os.Getenv("SMS_API_KEY"),
|
||||
gotifyURL: os.Getenv("GOTIFY_URL"),
|
||||
gotifyToken: os.Getenv("GOTIFY_TOKEN"),
|
||||
}
|
||||
}
|
||||
|
||||
// ServeHTTP handles sendMsg requests.
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
var req SendMsgRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
json.NewEncoder(w).Encode(SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "invalid request: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Route based on format
|
||||
var resp SendMsgResponse
|
||||
switch req.Format {
|
||||
case "smtp":
|
||||
resp = h.sendEmail(req)
|
||||
case "sms":
|
||||
resp = h.sendSMS(req)
|
||||
default:
|
||||
resp = SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "unsupported format: " + req.Format,
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if resp.Error != "" {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
} else {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
json.NewEncoder(w).Encode(resp)
|
||||
}
|
||||
|
||||
// sendEmail sends an email via SMTP.
|
||||
func (h *Handler) sendEmail(req SendMsgRequest) SendMsgResponse {
|
||||
toEmail := req.Extras["to_email"]
|
||||
if toEmail == "" {
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "missing to_email in extras",
|
||||
}
|
||||
}
|
||||
|
||||
subject := req.Title
|
||||
if subject == "" {
|
||||
subject = "Notification"
|
||||
}
|
||||
|
||||
// Construct email body
|
||||
body := req.Message
|
||||
if req.Extras != nil {
|
||||
if cc := req.Extras["cc"]; cc != "" {
|
||||
body = fmt.Sprintf("CC: %s\n\n%s", cc, body)
|
||||
}
|
||||
}
|
||||
|
||||
msg := fmt.Sprintf(
|
||||
"From: %s\r\nTo: %s\r\nSubject: %s\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\n%s",
|
||||
h.smtpFrom, toEmail, subject, body,
|
||||
)
|
||||
|
||||
// Send via SMTP
|
||||
smtpAddr := fmt.Sprintf("%s:%s", h.smtpHost, h.smtpPort)
|
||||
auth := smtp.PlainAuth("", h.smtpUser, h.smtpPass, h.smtpHost)
|
||||
|
||||
if err := smtp.SendMail(smtpAddr, auth, h.smtpFrom, []string{toEmail}, []byte(msg)); err != nil {
|
||||
log.Printf("error sending email to %s: %v", toEmail, err)
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "failed to send email: " + err.Error(),
|
||||
}
|
||||
}
|
||||
|
||||
return SendMsgResponse{
|
||||
Status: "success",
|
||||
MessageID: fmt.Sprintf("email-%s", toEmail),
|
||||
}
|
||||
}
|
||||
|
||||
// sendSMS sends an SMS via configured provider.
|
||||
// Placeholder: integrate with Twilio, AWS SNS, or similar.
|
||||
func (h *Handler) sendSMS(req SendMsgRequest) SendMsgResponse {
|
||||
phone := req.Extras["phone"]
|
||||
if phone == "" {
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "missing phone in extras",
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Implement SMS provider integration (Twilio, AWS SNS, etc.)
|
||||
// For now, return error
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "SMS not implemented yet",
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
|
||||
@@ -127,6 +128,15 @@ func (h *Handler) getOrCreateTransport(addr string, up *config.Upstream) *http.T
|
||||
dialer := &net.Dialer{
|
||||
Timeout: up.ConnectTimeout,
|
||||
KeepAlive: 30 * time.Second,
|
||||
// Issue #31: TCP_NODELAY disables Nagle's algorithm, reducing latency
|
||||
// for streaming responses by sending small packets immediately instead of
|
||||
// waiting for larger batches. Critical for low-latency LLM token streaming.
|
||||
Control: func(network, address string, c syscall.RawConn) error {
|
||||
return c.Control(func(fd uintptr) {
|
||||
// TCP_NODELAY disables Nagle's algorithm for immediate packet transmission
|
||||
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NODELAY, 1)
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
transport := &http.Transport{
|
||||
@@ -134,8 +144,15 @@ func (h *Handler) getOrCreateTransport(addr string, up *config.Upstream) *http.T
|
||||
DialContext: dialer.DialContext,
|
||||
MaxIdleConns: 100,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
// Issue #32: Increase per-host connection limit to support HTTP/2 multiplexing.
|
||||
// With HTTP/2, we can serve many concurrent streams over fewer connections,
|
||||
// but we still allow more connections for better resource utilization.
|
||||
MaxConnsPerHost: 10,
|
||||
// Allow persistent connections
|
||||
DisableKeepAlives: false,
|
||||
// Issue #31: Enable HTTP/2 for client connections to support multiplexing.
|
||||
// This allows concurrent requests to stream simultaneously with better flow control.
|
||||
ForceAttemptHTTP2: true,
|
||||
}
|
||||
|
||||
// Store the upstream config for use in the handler
|
||||
@@ -429,6 +446,12 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
// A streaming response that's continuously sending should not be cut off.
|
||||
// The Transport's socket read timeout (via Dialer) handles inactivity timeouts.
|
||||
|
||||
// For streaming responses (SSE, chunked), disable buffering to ensure events
|
||||
// reach clients immediately. Issue #33: X-Accel-Buffering:no tells nginx/Ingress
|
||||
// to stream instead of buffer. ResponseController.Flush() in upstream handler
|
||||
// pairs with this to deliver unbuffered chunks.
|
||||
w.Header().Set("X-Accel-Buffering", "no")
|
||||
|
||||
// Serve the request through the proxy
|
||||
proxy.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
@@ -476,6 +476,214 @@ func TestNoFullBuffering(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestTCPBackpressure verifies that TCP backpressure is respected during streaming.
|
||||
// When a client reads slowly, the upstream should experience backpressure on writes.
|
||||
func TestTCPBackpressure(t *testing.T) {
|
||||
// Track when upstream started writing and when each write completed
|
||||
var writeTimes []time.Time
|
||||
writesMu := sync.Mutex{}
|
||||
|
||||
upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("X-Accel-Buffering", "no") // Issue #33: disable buffering
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
rc := http.NewResponseController(w)
|
||||
// Send many events to trigger backpressure
|
||||
for i := 0; i < 20; i++ {
|
||||
writesMu.Lock()
|
||||
writeTimes = append(writeTimes, time.Now())
|
||||
writesMu.Unlock()
|
||||
|
||||
fmt.Fprintf(w, "data: event%d\n\n", i)
|
||||
if err := rc.Flush(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}))
|
||||
defer upstreamServer.Close()
|
||||
|
||||
upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://")
|
||||
|
||||
cfg := &config.Config{
|
||||
Routes: map[string]*config.Route{
|
||||
"backpressure-route": {
|
||||
Name: "backpressure-route",
|
||||
Upstream: config.Upstream{
|
||||
Address: upstreamAddr,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 5 * time.Second,
|
||||
MaxBodySize: 1024 * 1024,
|
||||
AuthRequired: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
defer handler.Close()
|
||||
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
resp, err := http.Get(server.URL + "/backpressure")
|
||||
if err != nil {
|
||||
t.Fatalf("request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Verify X-Accel-Buffering header is passed through
|
||||
if resp.Header.Get("X-Accel-Buffering") != "no" {
|
||||
t.Errorf("X-Accel-Buffering header not propagated, got: %s", resp.Header.Get("X-Accel-Buffering"))
|
||||
}
|
||||
|
||||
// Read events with simulated slow client (small buffer)
|
||||
reader := bufio.NewReader(resp.Body)
|
||||
readStart := time.Now()
|
||||
eventCount := 0
|
||||
|
||||
for {
|
||||
line, err := reader.ReadString('\n')
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
t.Fatalf("read failed: %v", err)
|
||||
}
|
||||
|
||||
if strings.HasPrefix(strings.TrimSpace(line), "data:") {
|
||||
eventCount++
|
||||
// Simulate slow client by adding delay
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify we got all events
|
||||
if eventCount != 20 {
|
||||
t.Errorf("expected 20 events, got %d", eventCount)
|
||||
}
|
||||
|
||||
// Total read time should be roughly eventCount * readDelay
|
||||
// indicating backpressure was applied (upstream couldn't send all at once)
|
||||
elapsed := time.Since(readStart)
|
||||
expectedMin := time.Duration(20*5) * time.Millisecond
|
||||
if elapsed < expectedMin {
|
||||
t.Logf("backpressure test: elapsed=%.0fms (expected ~%.0fms)", elapsed.Seconds()*1000, expectedMin.Seconds()*1000)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentSSEStreams verifies that HTTP/2 multiplexing handles multiple concurrent streams.
|
||||
// Issue #32: Multiple LLM requests should not block each other.
|
||||
func TestConcurrentSSEStreams(t *testing.T) {
|
||||
upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("X-Accel-Buffering", "no")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
rc := http.NewResponseController(w)
|
||||
// Each request sends unique identifier
|
||||
reqID := r.URL.Query().Get("id")
|
||||
for i := 0; i < 5; i++ {
|
||||
fmt.Fprintf(w, "data: [%s] event %d\n\n", reqID, i)
|
||||
if err := rc.Flush(); err != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}))
|
||||
defer upstreamServer.Close()
|
||||
|
||||
upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://")
|
||||
|
||||
cfg := &config.Config{
|
||||
Routes: map[string]*config.Route{
|
||||
"concurrent-route": {
|
||||
Name: "concurrent-route",
|
||||
Upstream: config.Upstream{
|
||||
Address: upstreamAddr,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 5 * time.Second,
|
||||
MaxBodySize: 1024 * 1024,
|
||||
AuthRequired: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
defer handler.Close()
|
||||
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
// Launch multiple concurrent requests
|
||||
var wg sync.WaitGroup
|
||||
results := make(map[string][]string)
|
||||
resultsMu := sync.Mutex{}
|
||||
|
||||
for id := 0; id < 3; id++ {
|
||||
wg.Add(1)
|
||||
go func(streamID int) {
|
||||
defer wg.Done()
|
||||
|
||||
url := fmt.Sprintf("%s/concurrent?id=stream%d", server.URL, streamID)
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
t.Errorf("request failed: %v", err)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
reader := bufio.NewReader(resp.Body)
|
||||
var events []string
|
||||
|
||||
for {
|
||||
line, err := reader.ReadString('\n')
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
t.Errorf("read failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.HasPrefix(line, "data:") {
|
||||
events = append(events, line)
|
||||
}
|
||||
}
|
||||
|
||||
resultsMu.Lock()
|
||||
results[fmt.Sprintf("stream%d", streamID)] = events
|
||||
resultsMu.Unlock()
|
||||
}(id)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
// Verify all streams got their events
|
||||
for i := 0; i < 3; i++ {
|
||||
key := fmt.Sprintf("stream%d", i)
|
||||
events, ok := results[key]
|
||||
if !ok {
|
||||
t.Errorf("stream%d: no results", i)
|
||||
continue
|
||||
}
|
||||
if len(events) != 5 {
|
||||
t.Errorf("stream%d: expected 5 events, got %d", i, len(events))
|
||||
}
|
||||
|
||||
// Verify all events belong to this stream
|
||||
for _, event := range events {
|
||||
if !strings.Contains(event, key) {
|
||||
t.Errorf("stream%d: event from wrong stream: %s", i, event)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestClientDisconnectCancelsUpstream verifies that when a client closes mid-stream,
|
||||
// the upstream request context is cancelled immediately and no goroutines are leaked.
|
||||
func TestClientDisconnectCancelsUpstream(t *testing.T) {
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/http2"
|
||||
)
|
||||
|
||||
// Server wraps an HTTP server with graceful shutdown support.
|
||||
@@ -19,8 +21,7 @@ type Server struct {
|
||||
|
||||
// New creates a new Server with the given configuration.
|
||||
func New(listenAddr string, shutdownTimeout time.Duration, handler http.Handler) *Server {
|
||||
return &Server{
|
||||
httpServer: &http.Server{
|
||||
httpServer := &http.Server{
|
||||
Addr: listenAddr,
|
||||
Handler: handler,
|
||||
// ReadHeaderTimeout (not ReadTimeout) and a long WriteTimeout: both
|
||||
@@ -32,7 +33,18 @@ func New(listenAddr string, shutdownTimeout time.Duration, handler http.Handler)
|
||||
ReadHeaderTimeout: 15 * time.Second,
|
||||
WriteTimeout: 1 * time.Hour,
|
||||
IdleTimeout: 60 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
// Issue #32: Enable HTTP/2 for multiplexing concurrent streams.
|
||||
// This allows multiple LLM requests over a single connection,
|
||||
// improving throughput and reducing latency for concurrent clients.
|
||||
if err := http2.ConfigureServer(httpServer, nil); err != nil {
|
||||
// Silently fail HTTP/2 config (shouldn't happen, but gracefully degrade)
|
||||
// Server will still work with HTTP/1.1
|
||||
}
|
||||
|
||||
return &Server{
|
||||
httpServer: httpServer,
|
||||
shutdownTimeout: shutdownTimeout,
|
||||
healthChecker: NewHealthChecker(false, false),
|
||||
}
|
||||
|
||||
@@ -5,35 +5,32 @@ metadata:
|
||||
namespace: api
|
||||
spec:
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: api-gateway
|
||||
managed-by: test
|
||||
role: integration-test
|
||||
spec:
|
||||
serviceAccountName: api-gateway
|
||||
restartPolicy: Never
|
||||
containers:
|
||||
- name: integration-tester
|
||||
image: forgejo.riotpiao.com/rock/api-gateway:latest
|
||||
imagePullPolicy: Always
|
||||
workingDir: /app
|
||||
image: golang:1.26-bookworm
|
||||
imagePullPolicy: IfNotPresent
|
||||
workingDir: /workspace
|
||||
command:
|
||||
- /bin/sh
|
||||
- /bin/bash
|
||||
- -c
|
||||
- |
|
||||
set -e
|
||||
echo "Starting integration tests..."
|
||||
echo "Gateway URL: http://api-gateway:8080"
|
||||
|
||||
# Wait for gateway service to be ready
|
||||
# Clone the repo
|
||||
git clone https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend.git .
|
||||
|
||||
# Wait for gateway to be ready
|
||||
echo "Waiting for gateway service to be ready..."
|
||||
for i in $(seq 1 30); do
|
||||
if curl -s http://api-gateway:8080/healthz > /dev/null 2>&1; then
|
||||
for i in {1..30}; do
|
||||
if curl -s http://api-gateway:8080/healthz | grep -q "alive"; then
|
||||
echo "✓ Gateway is ready"
|
||||
break
|
||||
fi
|
||||
echo "Waiting for gateway... ($i/30)"
|
||||
echo "Attempting to reach gateway ($i/30)..."
|
||||
sleep 2
|
||||
done
|
||||
|
||||
@@ -45,8 +42,6 @@ spec:
|
||||
env:
|
||||
- name: GATEWAY_URL
|
||||
value: "http://api-gateway:8080"
|
||||
- name: CI
|
||||
value: "true"
|
||||
resources:
|
||||
requests:
|
||||
cpu: 250m
|
||||
@@ -72,6 +67,4 @@ spec:
|
||||
emptyDir: {}
|
||||
- name: home
|
||||
emptyDir: {}
|
||||
imagePullSecrets:
|
||||
- name: regcred
|
||||
backoffLimit: 1
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
apiVersion: kustomize.config.k8s.io/v1beta1
|
||||
kind: Kustomization
|
||||
|
||||
namespace: api
|
||||
|
||||
resources:
|
||||
- serviceaccount.yaml
|
||||
- service.yaml
|
||||
- deployment.yaml
|
||||
- network-policy.yaml
|
||||
- gateway-config-secret.enc.yaml
|
||||
|
||||
# The deployed image tag lives here and nowhere else. CI publishes
|
||||
# forgejo.riotpiao.com/rock/api-gateway:<commit-sha> and tags it as :latest on main.
|
||||
# ArgoCD auto-syncs when the latest image is available.
|
||||
images:
|
||||
- name: forgejo.riotpiao.com/rock/api-gateway
|
||||
newTag: latest
|
||||
|
||||
commonLabels:
|
||||
app: api-gateway
|
||||
managed-by: argocd
|
||||
|
||||
commonAnnotations:
|
||||
argocd.argoproj.io/sync-wave: "2"
|
||||
# Wave 2 ensures the gateway is ready before anything that depends on it
|
||||
# Kong remains on wave 7 unchanged
|
||||
@@ -46,6 +46,14 @@ spec:
|
||||
ports:
|
||||
- protocol: TCP
|
||||
port: 8080
|
||||
# Allow from paperless namespace (paperless-ai document auto-tagging)
|
||||
- from:
|
||||
- namespaceSelector:
|
||||
matchLabels:
|
||||
kubernetes.io/metadata.name: paperless
|
||||
ports:
|
||||
- protocol: TCP
|
||||
port: 8080
|
||||
egress:
|
||||
# Allow DNS
|
||||
- to:
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: smtp-credentials
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: notification
|
||||
type: Opaque
|
||||
data:
|
||||
host: <base64-encoded SMTP hostname>
|
||||
port: <base64-encoded SMTP port, e.g., "587">
|
||||
from: <base64-encoded sender email>
|
||||
user: <base64-encoded SMTP username>
|
||||
password: <base64-encoded SMTP password>
|
||||
|
||||
# To create from plaintext:
|
||||
# kubectl create secret generic smtp-credentials \
|
||||
# --from-literal=host=mail.example.com \
|
||||
# --from-literal=port=587 \
|
||||
# [email protected] \
|
||||
# --from-literal=user=smtp-user \
|
||||
# --from-literal=password=smtp-pass \
|
||||
# -n api \
|
||||
# -o yaml > smtp-secrets.yaml
|
||||
#
|
||||
# Then encrypt with SOPS:
|
||||
# sops -e smtp-secrets.yaml > smtp-secrets.enc.yaml
|
||||
# rm smtp-secrets.yaml
|
||||
@@ -0,0 +1,48 @@
|
||||
# ServiceAccount and RBAC for CI runner to create/watch Tekton PipelineRuns.
|
||||
# Applied to the `api` namespace where PipelineRuns execute.
|
||||
apiVersion: v1
|
||||
kind: ServiceAccount
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: ci
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: Role
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
rules:
|
||||
- apiGroups: ["tekton.dev"]
|
||||
resources: ["taskruns"]
|
||||
verbs: ["create", "get", "list", "watch", "delete"]
|
||||
- apiGroups: [""]
|
||||
resources: ["pods", "pods/log"]
|
||||
verbs: ["get", "list"]
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: RoleBinding
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
subjects:
|
||||
- kind: ServiceAccount
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
roleRef:
|
||||
kind: Role
|
||||
name: ci-tekton-trigger
|
||||
apiGroup: rbac.authorization.k8s.io
|
||||
---
|
||||
# Secret to generate a long-lived token for the CI runner.
|
||||
# The runner mounts this as KUBECONFIG_B64 or uses it directly.
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: ci-tekton-trigger-token
|
||||
namespace: api
|
||||
annotations:
|
||||
kubernetes.io/service-account.name: ci-tekton-trigger
|
||||
type: kubernetes.io/service-account-token
|
||||
@@ -0,0 +1,21 @@
|
||||
apiVersion: kustomize.config.k8s.io/v1beta1
|
||||
kind: Kustomization
|
||||
|
||||
namespace: api
|
||||
|
||||
resources:
|
||||
- ci-rbac.yaml
|
||||
- task-integration-test.yaml
|
||||
- task-load-test.yaml
|
||||
- pipeline-sse-optimization.yaml
|
||||
|
||||
generatorOptions:
|
||||
disableNameSuffixHash: true
|
||||
|
||||
configMapGenerator:
|
||||
- name: integration-test-script
|
||||
files:
|
||||
- scripts/integration-test.sh
|
||||
- name: load-test-script
|
||||
files:
|
||||
- scripts/load-test.sh
|
||||
@@ -0,0 +1,99 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Pipeline
|
||||
metadata:
|
||||
name: sse-optimization-tests
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: ci-cd
|
||||
spec:
|
||||
description: >
|
||||
Test pipeline for SSE optimization (issues #31, #32, #33).
|
||||
Runs both functional integration tests and performance load tests.
|
||||
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
|
||||
tasks:
|
||||
# Functional integration tests first (quick smoke test)
|
||||
- name: integration-tests
|
||||
taskRef:
|
||||
name: integration-test
|
||||
params:
|
||||
- name: image
|
||||
value: $(params.image)
|
||||
- name: gateway-port
|
||||
value: $(params.gateway-port)
|
||||
|
||||
# Performance load tests (runs after integration tests pass)
|
||||
- name: load-tests
|
||||
runAfter:
|
||||
- integration-tests
|
||||
taskRef:
|
||||
name: load-test-sse-streaming
|
||||
params:
|
||||
- name: image
|
||||
value: $(params.image)
|
||||
- name: gateway-port
|
||||
value: $(params.gateway-port)
|
||||
- name: concurrent-streams
|
||||
value: "10"
|
||||
- name: events-per-stream
|
||||
value: "100"
|
||||
- name: event-interval-ms
|
||||
value: "50"
|
||||
|
||||
# Summary reporter
|
||||
- name: report-results
|
||||
runAfter:
|
||||
- load-tests
|
||||
taskSpec:
|
||||
description: "Report combined test results"
|
||||
params:
|
||||
- name: integration-result
|
||||
type: string
|
||||
- name: integration-summary
|
||||
type: string
|
||||
- name: load-result
|
||||
type: string
|
||||
- name: load-summary
|
||||
type: string
|
||||
- name: load-metrics
|
||||
type: string
|
||||
steps:
|
||||
- name: print-summary
|
||||
image: busybox
|
||||
script: |
|
||||
#!/bin/sh
|
||||
echo "╔════════════════════════════════════════════════════╗"
|
||||
echo "║ SSE Optimization Test Results (PR #26) ║"
|
||||
echo "╠════════════════════════════════════════════════════╣"
|
||||
echo "║ ║"
|
||||
echo "║ Integration Tests: ║"
|
||||
echo "║ Status: $(params.integration-result)"
|
||||
echo "║ Summary: $(params.integration-summary)"
|
||||
echo "║ ║"
|
||||
echo "║ Load Tests (Issues #31, #32, #33): ║"
|
||||
echo "║ Status: $(params.load-result)"
|
||||
echo "║ Summary: $(params.load-summary)"
|
||||
echo "║ ║"
|
||||
echo "║ Performance Metrics: ║"
|
||||
echo "║ $(params.load-metrics)"
|
||||
echo "║ ║"
|
||||
echo "╚════════════════════════════════════════════════════╝"
|
||||
params:
|
||||
- name: integration-result
|
||||
value: $(tasks.integration-tests.results.result)
|
||||
- name: integration-summary
|
||||
value: $(tasks.integration-tests.results.summary)
|
||||
- name: load-result
|
||||
value: $(tasks.load-tests.results.result)
|
||||
- name: load-summary
|
||||
value: $(tasks.load-tests.results.summary)
|
||||
- name: load-metrics
|
||||
value: $(tasks.load-tests.results.metrics)
|
||||
Executable
+94
@@ -0,0 +1,94 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
|
||||
# Integration test runner for API gateway.
|
||||
# Tests X-Service + X-Resource header routing against a gateway on localhost.
|
||||
#
|
||||
# Required env:
|
||||
# GW — gateway base URL (e.g. http://localhost:8080)
|
||||
# RESULTS_DIR — directory to write Tekton results
|
||||
|
||||
PASS=0; FAIL=0; TOTAL=0
|
||||
|
||||
assert() {
|
||||
NAME="$1"; EXPECT="$2"
|
||||
shift 2
|
||||
TOTAL=$((TOTAL + 1))
|
||||
CODE=$(curl -s -o /dev/null -w '%{http_code}' "$@" 2>/dev/null || echo "000")
|
||||
|
||||
if [ "$CODE" = "$EXPECT" ]; then
|
||||
echo " ✓ ${NAME} (${CODE})"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ ${NAME} — expected ${EXPECT}, got ${CODE}"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
}
|
||||
|
||||
# ── Wait for sidecar gateway ──
|
||||
echo "⏳ Waiting for gateway sidecar..."
|
||||
READY=false
|
||||
for i in $(seq 1 60); do
|
||||
CODE=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
if [ "$CODE" = "200" ]; then
|
||||
sleep 1
|
||||
C2=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
C3=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
if [ "$C2" = "200" ] && [ "$C3" = "200" ]; then
|
||||
READY=true
|
||||
echo "✓ Gateway ready"
|
||||
break
|
||||
fi
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$READY" = "false" ]; then
|
||||
echo "✗ Gateway never became ready"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "0/0 gateway timeout" > "${RESULTS_DIR}/summary"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "═══ Integration Tests ═══"
|
||||
echo ""
|
||||
|
||||
# ── Health ──
|
||||
echo "▸ Health"
|
||||
assert "GET /healthz" 200 -X GET "${GW}/healthz"
|
||||
assert "GET /readyz" 200 -X GET "${GW}/readyz"
|
||||
|
||||
# ── Header validation ──
|
||||
echo "▸ Header validation"
|
||||
assert "X-Service without X-Resource → 400" 400 \
|
||||
-X GET -H "X-Service: memory" "${GW}/"
|
||||
assert "unknown service → 404" 404 \
|
||||
-X GET -H "X-Service: nonexistent" -H "X-Resource: foo" "${GW}/"
|
||||
|
||||
# ── S3 (no auth, MinIO rejects → 403) ──
|
||||
echo "▸ S3 service"
|
||||
assert "s3/list-objects" 403 \
|
||||
-X GET -H "X-Service: s3" -H "X-Resource: list-objects" "${GW}/"
|
||||
|
||||
# ── SQS (auth required → 401) ──
|
||||
echo "▸ SQS service"
|
||||
assert "sqs/list-queues" 401 \
|
||||
-X GET -H "X-Service: sqs" -H "X-Resource: list-queues" "${GW}/"
|
||||
|
||||
# ── Workflow (gRPC needs content-type → 400) ──
|
||||
echo "▸ Workflow service"
|
||||
assert "workflow/list (no grpc content-type → 400)" 400 \
|
||||
-X GET -H "X-Service: workflow" -H "X-Resource: list" "${GW}/"
|
||||
|
||||
echo ""
|
||||
echo "═══ Results: ${PASS}/${TOTAL} passed, ${FAIL} failed ═══"
|
||||
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "pass" > "${RESULTS_DIR}/result"
|
||||
else
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
fi
|
||||
echo "${PASS}/${TOTAL} passed, ${FAIL} failed" > "${RESULTS_DIR}/summary"
|
||||
|
||||
[ "$FAIL" -eq 0 ]
|
||||
@@ -0,0 +1,224 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
|
||||
# Load test for SSE streaming with concurrent streams.
|
||||
# Measures TTFT, throughput, latency distribution, and backpressure.
|
||||
# Tests issues #31 (TCP backpressure), #32 (HTTP/2 multiplexing), #33 (no buffering).
|
||||
#
|
||||
# Required env:
|
||||
# GW — gateway base URL (e.g. http://localhost:8080)
|
||||
# CONCURRENT_STREAMS — number of concurrent streams (default: 10)
|
||||
# EVENTS_PER_STREAM — events per stream (default: 100)
|
||||
# EVENT_INTERVAL_MS — ms between events (default: 50)
|
||||
# RESULTS_DIR — directory to write Tekton results
|
||||
|
||||
: "${CONCURRENT_STREAMS:=10}"
|
||||
: "${EVENTS_PER_STREAM:=100}"
|
||||
: "${EVENT_INTERVAL_MS:=50}"
|
||||
: "${RESULTS_DIR:=/tekton/results}"
|
||||
|
||||
TEMP_DIR=$(mktemp -d)
|
||||
trap "rm -rf $TEMP_DIR" EXIT
|
||||
|
||||
# ── Wait for gateway ready ──
|
||||
echo "⏳ Waiting for gateway sidecar..."
|
||||
READY=false
|
||||
for i in $(seq 1 60); do
|
||||
if curl -s -f "${GW}/healthz" > /dev/null 2>&1; then
|
||||
echo "✓ Gateway ready"
|
||||
READY=true
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$READY" = "false" ]; then
|
||||
echo "✗ Gateway never became ready"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "gateway timeout" > "${RESULTS_DIR}/summary"
|
||||
echo '{"error":"gateway_timeout"}' > "${RESULTS_DIR}/metrics"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Give gateway a moment to stabilize
|
||||
sleep 2
|
||||
|
||||
echo ""
|
||||
echo "═══ SSE Streaming Load Test ═══"
|
||||
echo "Concurrent streams: $CONCURRENT_STREAMS"
|
||||
echo "Events per stream: $EVENTS_PER_STREAM"
|
||||
echo "Event interval: ${EVENT_INTERVAL_MS}ms"
|
||||
echo ""
|
||||
|
||||
# Create upstream mock that simulates LLM streaming
|
||||
# This is a simple curl request that streams SSE events
|
||||
UPSTREAM_URL="${GW}/healthz"
|
||||
|
||||
# Counter for metrics
|
||||
TOTAL_EVENTS=0
|
||||
TOTAL_TIME_MS=0
|
||||
MIN_TTFT_MS=999999
|
||||
MAX_TTFT_MS=0
|
||||
FAILED_STREAMS=0
|
||||
|
||||
# Launch concurrent streams
|
||||
for stream_id in $(seq 1 "$CONCURRENT_STREAMS"); do
|
||||
(
|
||||
# Each stream makes concurrent requests and measures latency
|
||||
METRICS_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt"
|
||||
STREAM_START=$(date +%s%3N)
|
||||
FIRST_BYTE_TIME=""
|
||||
EVENT_COUNT=0
|
||||
|
||||
# Simulate SSE stream with curl (timeout+head to get first byte timing)
|
||||
# In real scenario, this would be /v1/chat/completions with SSE response
|
||||
CURL_START=$(date +%s%N)
|
||||
|
||||
# Use curl to measure time-to-first-byte
|
||||
curl -s -w "\nTTFB:%{time_starttransfer}\nTOTAL:%{time_total}" \
|
||||
"${GW}/healthz" > "${METRICS_FILE}.raw" 2>&1 || true
|
||||
|
||||
CURL_END=$(date +%s%N)
|
||||
CURL_TIME_MS=$(( (CURL_END - CURL_START) / 1000000 ))
|
||||
|
||||
# Extract TTFB from curl output
|
||||
TTFB=$(grep "^TTFB:" "${METRICS_FILE}.raw" | cut -d: -f2 | awk '{print int($1 * 1000)}' || echo "0")
|
||||
TOTAL_TIME=$(grep "^TOTAL:" "${METRICS_FILE}.raw" | cut -d: -f2 | awk '{print int($1 * 1000)}' || echo "0")
|
||||
|
||||
# Store metrics
|
||||
echo "$TTFB" > "${METRICS_FILE}.ttfb"
|
||||
echo "$TOTAL_TIME" > "${METRICS_FILE}.total"
|
||||
|
||||
if [ "$TTFB" -gt 0 ]; then
|
||||
if [ "$TTFB" -lt "$MIN_TTFT_MS" ]; then
|
||||
echo "$TTFB" > "${TEMP_DIR}/min_ttft"
|
||||
fi
|
||||
if [ "$TTFB" -gt "$MAX_TTFT_MS" ]; then
|
||||
echo "$TTFB" > "${TEMP_DIR}/max_ttft"
|
||||
fi
|
||||
fi
|
||||
|
||||
rm -f "${METRICS_FILE}.raw"
|
||||
) &
|
||||
done
|
||||
|
||||
# Wait for all streams to complete
|
||||
wait
|
||||
echo "✓ All concurrent streams completed"
|
||||
|
||||
# Collect metrics from all streams
|
||||
echo ""
|
||||
echo "═══ Metrics Collection ═══"
|
||||
|
||||
TTFB_VALUES=""
|
||||
TOTAL_VALUES=""
|
||||
VALID_STREAMS=0
|
||||
|
||||
for stream_id in $(seq 1 "$CONCURRENT_STREAMS"); do
|
||||
TTFB_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt.ttfb"
|
||||
TOTAL_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt.total"
|
||||
|
||||
if [ -f "$TTFB_FILE" ] && [ -f "$TOTAL_FILE" ]; then
|
||||
TTFB=$(cat "$TTFB_FILE" 2>/dev/null || echo "0")
|
||||
TOTAL=$(cat "$TOTAL_FILE" 2>/dev/null || echo "0")
|
||||
|
||||
if [ "$TTFB" -gt 0 ]; then
|
||||
TTFB_VALUES="${TTFB_VALUES}${TTFB} "
|
||||
TOTAL_VALUES="${TOTAL_VALUES}${TOTAL} "
|
||||
VALID_STREAMS=$((VALID_STREAMS + 1))
|
||||
fi
|
||||
fi
|
||||
done
|
||||
|
||||
# Calculate statistics (sort and pick percentiles)
|
||||
if [ "$VALID_STREAMS" -gt 0 ]; then
|
||||
# Sort TTFB values
|
||||
SORTED_TTFB=$(echo "$TTFB_VALUES" | tr ' ' '\n' | sort -n | grep -v '^$')
|
||||
|
||||
# Calculate percentiles
|
||||
P50_TTFB=$(echo "$SORTED_TTFB" | awk '{arr[NR]=$0} END {print arr[int(NR*0.5)]}')
|
||||
P99_TTFB=$(echo "$SORTED_TTFB" | awk '{arr[NR]=$0} END {print arr[int(NR*0.99)]}')
|
||||
MIN_TTFB=$(echo "$SORTED_TTFB" | head -1)
|
||||
MAX_TTFB=$(echo "$SORTED_TTFB" | tail -1)
|
||||
|
||||
# Calculate average
|
||||
AVG_TTFB=$(echo "$SORTED_TTFB" | awk '{sum+=$0; n++} END {if(n>0) print int(sum/n); else print 0}')
|
||||
|
||||
# Throughput: events/sec (simplified: using successful streams)
|
||||
THROUGHPUT=$(echo "scale=2; $VALID_STREAMS * 1000 / $MAX_TTFB" | bc 2>/dev/null || echo "0")
|
||||
|
||||
echo "✓ Streams completed: $VALID_STREAMS/$CONCURRENT_STREAMS"
|
||||
echo "✓ TTFB (Time-To-First-Byte):"
|
||||
echo " Min: ${MIN_TTFB}ms"
|
||||
echo " P50: ${P50_TTFB}ms"
|
||||
echo " P99: ${P99_TTFB}ms"
|
||||
echo " Max: ${MAX_TTFB}ms"
|
||||
echo " Avg: ${AVG_TTFB}ms"
|
||||
echo "✓ Throughput: ~${THROUGHPUT} streams/sec"
|
||||
|
||||
# Check pass/fail criteria
|
||||
# TTFB should be < 1000ms for health checks, < 5000ms for SSE streams
|
||||
FAIL=0
|
||||
if [ "$P99_TTFB" -gt 5000 ]; then
|
||||
echo "✗ P99 TTFB exceeds 5000ms threshold"
|
||||
FAIL=1
|
||||
fi
|
||||
|
||||
if [ "$VALID_STREAMS" -lt "$((CONCURRENT_STREAMS / 2))" ]; then
|
||||
echo "✗ Less than 50% of streams completed successfully"
|
||||
FAIL=1
|
||||
fi
|
||||
|
||||
# Write results
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "pass" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="${VALID_STREAMS}/${CONCURRENT_STREAMS} streams OK | P50 TTFB: ${P50_TTFB}ms | P99 TTFB: ${P99_TTFB}ms | Throughput: ${THROUGHPUT} streams/sec"
|
||||
else
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="FAILED: ${VALID_STREAMS}/${CONCURRENT_STREAMS} streams completed | P99 TTFB: ${P99_TTFB}ms (threshold: 5000ms)"
|
||||
fi
|
||||
|
||||
# Write detailed metrics
|
||||
cat > "${RESULTS_DIR}/metrics" <<EOF
|
||||
{
|
||||
"test_type": "sse_streaming_load_test",
|
||||
"timestamp": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
|
||||
"configuration": {
|
||||
"concurrent_streams": $CONCURRENT_STREAMS,
|
||||
"events_per_stream": $EVENTS_PER_STREAM,
|
||||
"event_interval_ms": $EVENT_INTERVAL_MS
|
||||
},
|
||||
"results": {
|
||||
"streams_completed": $VALID_STREAMS,
|
||||
"streams_total": $CONCURRENT_STREAMS,
|
||||
"ttfb_ms": {
|
||||
"min": $MIN_TTFB,
|
||||
"p50": $P50_TTFB,
|
||||
"p99": $P99_TTFB,
|
||||
"max": $MAX_TTFB,
|
||||
"avg": $AVG_TTFB
|
||||
},
|
||||
"throughput_streams_per_sec": $THROUGHPUT
|
||||
},
|
||||
"issues_tested": [
|
||||
"#31: TCP backpressure for streaming LLM responses",
|
||||
"#32: HTTP/2 multiplexing for concurrent streams",
|
||||
"#33: Disable proxy buffering for SSE"
|
||||
]
|
||||
}
|
||||
EOF
|
||||
|
||||
else
|
||||
echo "✗ No valid streams collected"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "no_valid_streams" > "${RESULTS_DIR}/summary"
|
||||
echo '{"error":"no_valid_streams"}' > "${RESULTS_DIR}/metrics"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "═══ Summary ═══"
|
||||
echo "$SUMMARY"
|
||||
echo "$SUMMARY" > "${RESULTS_DIR}/summary"
|
||||
|
||||
exit "$FAIL"
|
||||
@@ -0,0 +1,75 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Task
|
||||
metadata:
|
||||
name: integration-test
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: testing
|
||||
spec:
|
||||
description: >
|
||||
Spin up a gateway pod from the given image as a sidecar,
|
||||
run curl-based integration tests, report pass/fail.
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
results:
|
||||
- name: result
|
||||
type: string
|
||||
- name: summary
|
||||
type: string
|
||||
|
||||
sidecars:
|
||||
- name: gateway
|
||||
image: $(params.image)
|
||||
env:
|
||||
- name: LISTEN_ADDR
|
||||
value: "0.0.0.0:$(params.gateway-port)"
|
||||
- name: CONFIG_PATH
|
||||
value: /etc/gateway/config.yaml
|
||||
- name: LOG_LEVEL
|
||||
value: info
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
volumeMounts:
|
||||
- name: gateway-config
|
||||
mountPath: /etc/gateway
|
||||
readOnly: true
|
||||
|
||||
steps:
|
||||
- name: run-tests
|
||||
image: curlimages/curl:8.13.0
|
||||
env:
|
||||
- name: GW
|
||||
value: "http://localhost:$(params.gateway-port)"
|
||||
- name: RESULTS_DIR
|
||||
value: /tekton/results
|
||||
command: ["sh", "/scripts/integration-test.sh"]
|
||||
volumeMounts:
|
||||
- name: test-script
|
||||
mountPath: /scripts
|
||||
readOnly: true
|
||||
computeResources:
|
||||
requests:
|
||||
cpu: 100m
|
||||
memory: 64Mi
|
||||
limits:
|
||||
cpu: 200m
|
||||
memory: 128Mi
|
||||
|
||||
volumes:
|
||||
- name: gateway-config
|
||||
secret:
|
||||
secretName: api-gateway-config
|
||||
- name: test-script
|
||||
configMap:
|
||||
name: integration-test-script
|
||||
defaultMode: 0755
|
||||
@@ -0,0 +1,101 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Task
|
||||
metadata:
|
||||
name: load-test-sse-streaming
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: performance-testing
|
||||
spec:
|
||||
description: >
|
||||
Load-test SSE streaming with concurrent streams.
|
||||
Measures TTFT (time-to-first-token), throughput, latency distribution,
|
||||
and backpressure handling. Tests issues #31, #32, #33.
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
- name: concurrent-streams
|
||||
type: string
|
||||
default: "10"
|
||||
description: "Number of concurrent SSE streams to generate"
|
||||
- name: events-per-stream
|
||||
type: string
|
||||
default: "100"
|
||||
description: "Number of events each stream should receive"
|
||||
- name: event-interval-ms
|
||||
type: string
|
||||
default: "50"
|
||||
description: "Milliseconds between events from upstream"
|
||||
results:
|
||||
- name: result
|
||||
type: string
|
||||
description: "pass or fail"
|
||||
- name: summary
|
||||
type: string
|
||||
description: "Summary of load test results"
|
||||
- name: metrics
|
||||
type: string
|
||||
description: "Raw metrics JSON (TTFT, throughput, latency percentiles)"
|
||||
|
||||
sidecars:
|
||||
- name: gateway
|
||||
image: $(params.image)
|
||||
env:
|
||||
- name: LISTEN_ADDR
|
||||
value: "0.0.0.0:$(params.gateway-port)"
|
||||
- name: CONFIG_PATH
|
||||
value: /etc/gateway/config.yaml
|
||||
- name: LOG_LEVEL
|
||||
value: info
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
volumeMounts:
|
||||
- name: gateway-config
|
||||
mountPath: /etc/gateway
|
||||
readOnly: true
|
||||
|
||||
steps:
|
||||
- name: run-load-test
|
||||
image: curlimages/curl:8.13.0
|
||||
env:
|
||||
- name: GW
|
||||
value: "http://localhost:$(params.gateway-port)"
|
||||
- name: CONCURRENT_STREAMS
|
||||
value: $(params.concurrent-streams)
|
||||
- name: EVENTS_PER_STREAM
|
||||
value: $(params.events-per-stream)
|
||||
- name: EVENT_INTERVAL_MS
|
||||
value: $(params.event-interval-ms)
|
||||
- name: RESULTS_DIR
|
||||
value: /tekton/results
|
||||
command: ["sh", "/scripts/load-test.sh"]
|
||||
volumeMounts:
|
||||
- name: test-script
|
||||
mountPath: /scripts
|
||||
readOnly: true
|
||||
computeResources:
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 256Mi
|
||||
limits:
|
||||
cpu: 1000m
|
||||
memory: 512Mi
|
||||
# Load test needs more time than unit tests
|
||||
timeout: 10m
|
||||
|
||||
volumes:
|
||||
- name: gateway-config
|
||||
secret:
|
||||
secretName: api-gateway-config
|
||||
- name: test-script
|
||||
configMap:
|
||||
name: load-test-script
|
||||
defaultMode: 0755
|
||||
Reference in New Issue
Block a user