GKE Integration — Service-to-Service Task Dispatch

Tại Sao Quan Trọng

Cloud Tasks targets không phải chỉ external endpoints. Bạn có thể dispatch tasks tới GKE services inside cluster:

Application → Cloud Tasks Queue → GKE Service
  (create task)         (dispatch)     (handler pod)

Lợi ích:

• Rate control — don't overwhelm GKE backend service
• Retry semantics — automatic retry, exponential backoff
• Deduplication — prevent duplicate work inside cluster
• Observability — Cloud Tasks logging, metrics

---

Architecture

Typical Pattern

┌──────────────────────────────────────────────────────────────┐
│                    GCP Project                               │
├──────────────────────────────────────────────────────────────┤
│                                                               │
│  Application                Cloud Tasks              GKE     │
│  ┌─────────────────┐       ┌────────────┐      ┌──────────┐ │
│  │                 │       │            │      │          │ │
│  │ API Handler     │◄─────►│  Queue     │──────►│ Service  │ │
│  │                 │       │            │      │ Handler  │ │
│  │ CreateTask()    │       │            │      │          │ │
│  │                 │       │            │      └──────────┘ │
│  └─────────────────┘       └────────────┘                    │
│        (http)                   (rate)          (rate-limited)│
│                                                               │
└──────────────────────────────────────────────────────────────┘

Network Flow

1. Application (Cloud Run / App Engine)
   → CreateTask(url="https://gke-service.internal/webhook")

2. Cloud Tasks Queue (managed)
   → Waits for schedule_time

3. Dispatch Engine (Google-managed)
   → GET ID token for service account
   → Make HTTP request to GKE service

4. GKE Service (inside cluster)
   → Load Balancer routes to pod
   → Pod receives request (with auth headers)
   → Process task

5. Pod Response
   → Return 200-299 (success)
   → Cloud Tasks marks task completed
   → Task deleted

---

Setup: Service Account & Workload Identity

Step 1: Create Kubernetes Service Account (KSA)

apiVersion: v1
kind: ServiceAccount
metadata:
  name: cloud-tasks-handler
  namespace: default

kubectl apply -f sa.yaml

Step 2: Create Google Service Account (GSA)

gcloud iam service-accounts create gke-tasks-handler \
    --project=my-project \
    --display-name="GKE Tasks Handler"

Step 3: Workload Identity Binding

# Allow KSA to impersonate GSA
gcloud iam service-accounts add-iam-policy-binding \
    gke-tasks-handler@my-project.iam.gserviceaccount.com \
    --role=roles/iam.workloadIdentityUser \
    --member=serviceAccount:my-project.svc.id.goog[default/cloud-tasks-handler]

Step 4: Annotate KSA

apiVersion: v1
kind: ServiceAccount
metadata:
  name: cloud-tasks-handler
  namespace: default
  annotations:
    iam.gke.io/gcp-service-account: gke-tasks-handler@my-project.iam.gserviceaccount.com

Step 5: Configure Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: task-handler
  namespace: default
spec:
  replicas: 3
  selector:
    matchLabels:
      app: task-handler
  template:
    metadata:
      labels:
        app: task-handler
    spec:
      serviceAccountName: cloud-tasks-handler
      containers:
      - name: handler
        image: gcr.io/my-project/task-handler:v1
        ports:
        - containerPort: 8080

---

Create Tasks Targeting GKE Service

Service Endpoint

# Service inside GKE cluster
kubectl get svc
# NAME              TYPE           CLUSTER-IP       EXTERNAL-IP
# task-handler      LoadBalancer   10.0.1.100       35.123.45.67

For Cloud Tasks to reach it, service needs public IP (LoadBalancer type).

CreateTask Code

import cloudtasks "cloud.google.com/go/cloudtasks/apiv2"

func DispatchTaskToGKE(ctx context.Context, paymentID string) error {
    client := cloudtasks.NewClient(ctx)
    defer client.Close()
    
    req := &taskspb.CreateTaskRequest{
        Parent: fmt.Sprintf(
            "projects/%s/locations/%s/queues/%s",
            "my-project",
            "us-central1",
            "payment-tasks",
        ),
        Task: &taskspb.Task{
            HttpRequest: &taskspb.HttpRequest{
                Url: "https://task-handler-gke-service.example.com/process",
                HttpMethod: taskspb.HttpMethod_POST,
                Headers: map[string]string{
                    "Content-Type": "application/json",
                },
                Body: []byte(fmt.Sprintf(`{"payment_id": "%s"}`, paymentID)),
                OidcToken: &taskspb.OidcToken{
                    ServiceAccountEmail: "gke-tasks-handler@my-project.iam.gserviceaccount.com",
                    Audience: "https://task-handler-gke-service.example.com",
                },
            },
        },
    }
    
    _, err := client.CreateTask(ctx, req)
    return err
}

---

Handler Implementation (GKE Pod)

Receive & Validate Task

package main

import (
    "context"
    "encoding/json"
    "net/http"
    "strings"
    
    "google.golang.org/api/idtoken"
)

func TaskHandler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    
    // Extract auth token
    authHeader := r.Header.Get("Authorization")
    if authHeader == "" {
        http.Error(w, "Missing authorization", http.StatusUnauthorized)
        return
    }
    
    bearerToken := strings.TrimPrefix(authHeader, "Bearer ")
    
    // Validate OIDC token
    payload, err := idtoken.Validate(ctx, bearerToken, "https://task-handler-gke-service.example.com")
    if err != nil {
        http.Error(w, "Invalid token", http.StatusUnauthorized)
        return
    }
    
    // Verify service account
    if payload.Claims["email"] != "gke-tasks-handler@my-project.iam.gserviceaccount.com" {
        http.Error(w, "Unauthorized service account", http.StatusForbidden)
        return
    }
    
    // Extract task metadata
    taskName := r.Header.Get("X-CloudTasks-TaskName")
    retryCount := r.Header.Get("X-CloudTasks-TaskRetryCount")
    
    // Parse payload
    var taskReq struct {
        PaymentID string `json:"payment_id"`
    }
    json.NewDecoder(r.Body).Decode(&taskReq)
    
    // Log with context
    log.Printf("Processing task: %s (retry: %s, payment_id: %s)", taskName, retryCount, taskReq.PaymentID)
    
    // Process
    err = processPayment(ctx, taskReq.PaymentID)
    if err != nil {
        log.Printf("Error processing payment: %v", err)
        w.WriteHeader(http.StatusInternalServerError)
        json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
        return
    }
    
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(map[string]string{"status": "success"})
}

func main() {
    http.HandleFunc("/process", TaskHandler)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

---

Service-to-Service Patterns

Pattern 1: Request-Response with Rate Limiting

Service A (fast)          Service B (slow)
  Process 100 events       Process 10 events/sec
       ↓
  Create 100 tasks
       ↓
Cloud Tasks Queue (rate limit: 10 QPS)
       ↓
  Dispatch 10/sec to Service B

Benefit: Service B doesn't get overwhelmed by Service A spikes.

// Service A creates tasks
for i := 0; i < 100; i++ {
    dispatchTaskToGKE(ctx, event{...})
}

// Service B handles 10 QPS (rate-limited)
// Service A doesn't need to know Service B capacity

Pattern 2: Workflow Orchestration

Order → Payment Task → Email Task → Inventory Task
  A          (rate-limited)
               ↓
           Error?
               ↓
          Retry Task

Each step is explicit task, can be rate-limited independently

// Step 1: Charge payment
taskID := fmt.Sprintf("order-%s-charge", orderID)
dispatchTaskToGKE(ctx, "payment-service", taskID, orderData)

// Step 2 (executed after Step 1 completes): Send email
// Configured in Task 1's handler to create Task 2
func PaymentHandler(w http.ResponseWriter, r *http.Request) {
    err := chargePayment()
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    
    // Create next task
    dispatchTaskToGKE(ctx, "email-service", taskID2, emailData)
    
    w.WriteHeader(http.StatusOK)
}

Pattern 3: Fanout from External Service

Cloud Function          Cloud Tasks         GKE Services
  (HTTP trigger)        (fan-out)           (parallel)
       ↓
  Image uploaded
       ↓
  Create 3 tasks:
    - Compress
    - Generate thumbnail
    - Index metadata
       ↓
   Dispatch parallel
       ↓
  Service 1, 2, 3 all running

---

Network Setup

GKE Service Exposure

# Internal Service (not exposed)
apiVersion: v1
kind: Service
metadata:
  name: task-handler
spec:
  type: ClusterIP  # Only internal
  ports:
  - port: 8080

# OR: External Service (exposed to Cloud Tasks)
apiVersion: v1
kind: Service
metadata:
  name: task-handler
spec:
  type: LoadBalancer
  ports:
  - port: 443
    targetPort: 8080
  annotations:
    cloud.google.com/load-balancer-type: "External"

Network Security

# Only allow traffic from Cloud Tasks
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-cloud-tasks
spec:
  podSelector:
    matchLabels:
      app: task-handler
  policyTypes:
  - Ingress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: cloud-tasks-sources  # Tag for Cloud Tasks IPs
    ports:
    - protocol: TCP
      port: 8080

Firewall Rules

# Allow Cloud Tasks to reach GKE load balancer
gcloud compute firewall-rules create allow-cloud-tasks \
    --direction=INGRESS \
    --priority=1000 \
    --source-ranges=0.0.0.0/0 \
    --allow=tcp:443 \
    --target-tags=gke-nodes

---

Troubleshooting

Task Not Reaching Handler

Checklist:
  1. Service has public LoadBalancer IP?
     kubectl get svc task-handler
     
  2. OIDC token valid?
     Check service account setup
     
  3. Network connectivity?
     gcloud compute security-policies rules describe ...
     
  4. Handler listening?
     kubectl logs -f deployment/task-handler

High Latency

Likely causes:
  1. Pod too slow → add replicas
  2. Network latency (inter-zone) → use regional service
  3. Rate limiting → check Cloud Tasks queue config
  
Solution:
  - kubectl scale deployment task-handler --replicas=10
  - kubectl get hpa (check autoscaling)
  - Increase queue max_dispatches_per_second

Authentication Failures

Errors:
  "Invalid token" → OIDC token validation failed
    → Check idtoken.Validate logic
    → Verify audience matches
    
  "Unauthorized service account" → Wrong GSA
    → Check service account email in request
    → Verify workload identity binding

Debug:
  kubectl describe pod <pod-name>
  → Check if GOOGLE_APPLICATION_CREDENTIALS set
  → Check if workload identity properly configured

---

Performance Considerations

Handler Capacity

GKE Node: 4 CPU, 16 GB RAM
Pod: 1 CPU, 2 GB RAM

Pod throughput:
  - 10 simple requests/sec
  - 5 complex requests/sec

Config Cloud Tasks:
  max_dispatches_per_second: 5 (conservative)
  max_concurrent_dispatches: 20 (buffer for latency)
  max_burst_size: 5

Scaling Handler

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: task-handler
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: task-handler
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

---

References

• Workload Identity on GKE
• GKE Service Exposure
• Cloud Tasks HTTP Targets
• OIDC Token Validation