Managing App Lifecycles with Kubernetes

Flyte manages the lifecycle of applications by orchestrating Knative Services (KServices) within a dedicated Kubernetes namespace. The AppK8sClient in app/internal/k8s/app_client.go serves as the primary interface for deploying, scaling, and monitoring these applications.

Deploying and Updating Apps

To deploy an application, Flyte constructs a Knative Service manifest and applies it to the cluster. The deployment process is idempotent; it uses a SHA256 hash of the application specification to determine if an update is necessary.

// From app/internal/k8s/app_client.go

func (c *AppK8sClient) Deploy(ctx context.Context, app *flyteapp.App) error {
	appID := app.GetMetadata().GetId()
	ns := AppNamespace // "flyte"
	name := KServiceName(appID)

	// Ensure the target namespace exists
	if err := k8s.EnsureNamespaceExists(ctx, c.k8sClient, ns); err != nil {
		return fmt.Errorf("failed to ensure namespace %s: %w", ns, err)
	}

	ksvc, err := c.buildKService(app)
	if err != nil {
		return fmt.Errorf("failed to build KService for app %s: %w", name, err)
	}

	existing := &servingv1.Service{}
	err = c.k8sClient.Get(ctx, client.ObjectKey{Name: name, Namespace: ns}, existing)
	
	// Create if it doesn't exist
	if k8serrors.IsNotFound(err) {
		return c.k8sClient.Create(ctx, ksvc)
	}

	// Idempotency check: skip update if the spec SHA matches
	// unless the app is currently in a stopped state.
	existingStopped := existing.Labels != nil && existing.Labels["flyte.org/app-stopped"] == "true"
	if !existingStopped && existing.Annotations["flyte.org/spec-sha"] == ksvc.Annotations["flyte.org/spec-sha"] {
		return nil
	}

	// Update existing service
	existing.Spec = ksvc.Spec
	// ... (merge labels and annotations)
	return c.k8sClient.Update(ctx, existing)
}

Deterministic Naming

Flyte generates unique, DNS-compliant names for KServices using the KServiceName helper. Because Kubernetes DNS labels are limited to 63 characters, Flyte uses a deterministic 8-character SHA256 suffix if the combined name-project-domain string exceeds the limit.

func KServiceName(id *flyteapp.Identifier) string {
	raw := strings.ToLower(fmt.Sprintf("%s-%s-%s", id.GetName(), id.GetProject(), id.GetDomain()))
	if len(raw) <= 63 {
		return raw
	}
	// Fallback for long names to ensure uniqueness within 63 chars
	sum := sha256.Sum256([]byte(id.GetProject() + "/" + id.GetDomain() + "/" + id.GetName()))
	suffix := hex.EncodeToString(sum[:4])
	return raw[:54] + "-" + suffix
}

Stopping Apps (Scaling to Zero)

Stopping an app in Flyte does not delete the Kubernetes resource. Instead, it scales the deployment to zero and restricts its visibility to cluster-local to prevent external traffic from reaching it.

To ensure immediate pod termination, Flyte patches the KService and explicitly deletes the LatestReadyRevision. Without this deletion, pods might persist until the Knative "stable window" expires.

func (c *AppK8sClient) Stop(ctx context.Context, appID *flyteapp.Identifier) error {
	ns := AppNamespace
	name := KServiceName(appID)
	
	// Patch to scale to zero and make cluster-local
	patch := []byte(fmt.Sprintf(
		`{"metadata":{"labels":{"flyte.org/app-stopped":"true","networking.knative.dev/visibility":"cluster-local"}},"spec":{"template":{"metadata":{"annotations":{"autoscaling.knative.dev/min-scale":"0","autoscaling.knative.dev/initial-scale":"0"}}}}}`,
	))
	
	ksvc := &servingv1.Service{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}}
	if err := c.k8sClient.Patch(ctx, ksvc, client.RawPatch(types.MergePatchType, patch)); err != nil {
		return err
	}

	// Force immediate termination by deleting the latest revision
	current := &servingv1.Service{}
	if err := c.k8sClient.Get(ctx, client.ObjectKey{Name: name, Namespace: ns}, current); err == nil {
		if revName := current.Status.LatestReadyRevisionName; revName != "" {
			rev := &servingv1.Revision{ObjectMeta: metav1.ObjectMeta{Name: revName, Namespace: ns}}
			_ = c.k8sClient.Delete(ctx, rev)
		}
	}
	return nil
}

Internal Networking and Service Discovery

Flyte facilitates service-to-service communication by injecting an INTERNAL_APP_ENDPOINT_PATTERN environment variable into every application container. This allows apps to resolve other Flyte apps within the same cluster using a predictable URL format.

The pattern is constructed in buildKService using the NamespacedNameSuffixTemplate configuration:

// In app/internal/k8s/app_client.go

suffix := renderNamespacedSuffix(c.cfg.NamespacedNameSuffixTemplate, appID.GetProject(), appID.GetDomain())
podSpec.Containers[0].Env = append(podSpec.Containers[0].Env, corev1.EnvVar{
    Name:  "INTERNAL_APP_ENDPOINT_PATTERN",
    Value: fmt.Sprintf("http://{app_fqdn}-%s.%s.svc.cluster.local", suffix, ns),
})

Monitoring App Status

Flyte uses a Kubernetes informer to watch for changes to KServices. The AppK8sClient maintains a subscription model where internal services can listen for status updates (e.g., when an app transitions from Deploying to Ready).

The handleKServiceEvent method translates Knative events into Flyte WatchResponse messages:

func (c *AppK8sClient) handleKServiceEvent(ctx context.Context, ksvc *servingv1.Service, eventType k8swatch.EventType) {
	app, err := c.kserviceToApp(ctx, ksvc)
	if err != nil {
		return
	}

	var resp *flyteapp.WatchResponse
	switch eventType {
	case k8swatch.Added:
		resp = &flyteapp.WatchResponse{Event: &flyteapp.WatchResponse_CreateEvent{...}}
	case k8swatch.Modified:
		resp = &flyteapp.WatchResponse{Event: &flyteapp.WatchResponse_UpdateEvent{...}}
	case k8swatch.Deleted:
		resp = &flyteapp.WatchResponse{Event: &flyteapp.WatchResponse_DeleteEvent{...}}
	}

	c.notifySubscribers(ctx, app.GetMetadata().GetId().GetName(), resp)
}

Status Mapping

The application status is derived from Knative's ServiceStatus conditions. Flyte maps these conditions to its internal state machine, identifying if the app is READY, FAILED, or STOPPED based on the presence of the flyte.org/app-stopped label and the Ready condition of the KService.