kserve/pkg/controller/v1alpha1/inferencegraph/knative_reconciler.go

/*
Copyright 2022 The KServe Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package inferencegraph

import (
	"context"
	"encoding/json"
	"reflect"
	"strings"

	"github.com/pkg/errors"
	"google.golang.org/protobuf/proto"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierr "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"knative.dev/pkg/kmp"
	knserving "knative.dev/serving/pkg/apis/serving"
	knservingv1 "knative.dev/serving/pkg/apis/serving/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/kserve/kserve/pkg/apis/serving/v1alpha1"
	"github.com/kserve/kserve/pkg/constants"
	knutils "github.com/kserve/kserve/pkg/controller/v1alpha1/utils"
	"github.com/kserve/kserve/pkg/utils"
)

var log = logf.Log.WithName("GraphKsvcReconciler")

type GraphKnativeServiceReconciler struct {
	client  client.Client
	scheme  *runtime.Scheme
	Service *knservingv1.Service
}

func NewGraphKnativeServiceReconciler(client client.Client,
	scheme *runtime.Scheme,
	ksvc *knservingv1.Service,
) *GraphKnativeServiceReconciler {
	return &GraphKnativeServiceReconciler{
		client:  client,
		scheme:  scheme,
		Service: ksvc,
	}
}

func reconcileKsvc(desired *knservingv1.Service, existing *knservingv1.Service) error {
	// Return if no differences to reconcile.
	if semanticEquals(desired, existing) {
		return nil
	}

	// Reconcile differences and update
	// knative mutator defaults the enableServiceLinks to false which would generate a diff despite no changes on desired knative service
	// https://github.com/knative/serving/blob/main/pkg/apis/serving/v1/revision_defaults.go#L134
	if desired.Spec.Template.Spec.EnableServiceLinks == nil &&
		existing.Spec.Template.Spec.EnableServiceLinks != nil &&
		!*existing.Spec.Template.Spec.EnableServiceLinks {
		desired.Spec.Template.Spec.EnableServiceLinks = proto.Bool(false)
	}
	diff, err := kmp.SafeDiff(desired.Spec.ConfigurationSpec, existing.Spec.ConfigurationSpec)
	if err != nil {
		return errors.Wrapf(err, "failed to diff inference graph knative service configuration spec")
	}
	log.Info("inference graph knative service configuration diff (-desired, +observed):", "diff", diff)
	existing.Spec.ConfigurationSpec = desired.Spec.ConfigurationSpec
	existing.Labels = desired.Labels
	existing.Spec.Traffic = desired.Spec.Traffic
	return nil
}

func (r *GraphKnativeServiceReconciler) Reconcile(ctx context.Context) (*knservingv1.ServiceStatus, error) {
	desired := r.Service
	existing := &knservingv1.Service{}

	forceStopRuntime := false
	if val, exist := desired.Spec.Template.Annotations[constants.StopAnnotationKey]; exist {
		forceStopRuntime = strings.EqualFold(val, "true")
	}

	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		log.Info("Updating inference graph knative service", "namespace", desired.Namespace, "name", desired.Name)
		if err := r.client.Get(ctx, types.NamespacedName{Name: desired.Name, Namespace: desired.Namespace}, existing); err != nil {
			return err
		}

		if forceStopRuntime {
			log.Info("Deleting inference graph knative service", "namespace", existing.Namespace, "name", existing.Name)
			if existing.GetDeletionTimestamp() == nil { // check if the ksvc was already deleted
				err := r.client.Delete(ctx, existing)
				if err != nil {
					return err
				}
			}
			return nil
		}

		// Set ResourceVersion which is required for update operation.
		desired.ResourceVersion = existing.ResourceVersion
		// Add immutable annotations to avoid validation error during dry-run update.
		desired.Annotations[knserving.CreatorAnnotation] = existing.Annotations[knserving.CreatorAnnotation]
		desired.Annotations[knserving.UpdaterAnnotation] = existing.Annotations[knserving.UpdaterAnnotation]

		// Do a dry-run update to avoid diffs generated by default values introduced by knative's defaulter webhook.
		// This will populate our local knative service object with any default values
		// that are present on the remote version.
		if err := r.client.Update(ctx, desired, client.DryRunAll); err != nil {
			// log only if it is not resource conflict error to avoid spamming
			if !apierr.IsConflict(err) {
				log.Error(err, "Failed to perform dry-run update of knative service", "service", desired.Name)
			}
			return err
		}

		if err := reconcileKsvc(desired, existing); err != nil {
			return err
		}
		return r.client.Update(ctx, existing)
	})
	if err != nil {
		// Create service if it does not exist
		if apierr.IsNotFound(err) {
			if !forceStopRuntime {
				log.Info("Creating inference graph knative service", "namespace", desired.Namespace, "name", desired.Name)
				return &desired.Status, r.client.Create(ctx, desired)
			}
			return &desired.Status, nil
		}
		return &existing.Status, errors.Wrapf(err, "fails to reconcile inference graph knative service")
	}
	return &existing.Status, nil
}

func semanticEquals(desiredService, service *knservingv1.Service) bool {
	return equality.Semantic.DeepEqual(desiredService.Spec.ConfigurationSpec, service.Spec.ConfigurationSpec) &&
		equality.Semantic.DeepEqual(desiredService.Labels, service.Labels) &&
		equality.Semantic.DeepEqual(desiredService.Spec.RouteSpec, service.Spec.RouteSpec)
}

func createKnativeService(
	componentMeta metav1.ObjectMeta,
	graph *v1alpha1.InferenceGraph,
	config *RouterConfig,
) *knservingv1.Service {
	bytes, err := json.Marshal(graph.Spec)
	if err != nil {
		return nil
	}
	annotations := componentMeta.GetAnnotations()
	if annotations == nil {
		annotations = make(map[string]string)
	}

	knutils.SetAutoScalingAnnotations(
		annotations,
		graph.Spec.ScaleTarget,
		(*string)(graph.Spec.ScaleMetric),
		graph.Spec.MinReplicas,
		graph.Spec.MaxReplicas,
		log,
	)

	// ksvc metadata.annotations
	ksvcAnnotations := make(map[string]string)

	if value, ok := annotations[constants.KnativeOpenshiftEnablePassthroughKey]; ok {
		ksvcAnnotations[constants.KnativeOpenshiftEnablePassthroughKey] = value
		delete(annotations, constants.KnativeOpenshiftEnablePassthroughKey)
	}

	labels := utils.Filter(componentMeta.Labels, func(key string) bool {
		return !utils.Includes(constants.RevisionTemplateLabelDisallowedList, key)
	})
	labels[constants.InferenceGraphLabel] = componentMeta.Name
	labels[constants.KServeWorkloadKind] = "InferenceGraph"
	service := &knservingv1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        componentMeta.Name,
			Namespace:   componentMeta.Namespace,
			Labels:      componentMeta.Labels,
			Annotations: ksvcAnnotations,
		},
		Spec: knservingv1.ServiceSpec{
			ConfigurationSpec: knservingv1.ConfigurationSpec{
				Template: knservingv1.RevisionTemplateSpec{
					ObjectMeta: metav1.ObjectMeta{
						Labels:      labels,
						Annotations: annotations,
					},
					Spec: knservingv1.RevisionSpec{
						TimeoutSeconds: graph.Spec.TimeoutSeconds,
						PodSpec: corev1.PodSpec{
							Containers: []corev1.Container{
								{
									Image:           config.Image,
									ImagePullPolicy: corev1.PullPolicy(config.ImagePullPolicy),
									Args: []string{
										"--graph-json",
										string(bytes),
									},
									Resources: constructResourceRequirements(*graph, *config),
									SecurityContext: &corev1.SecurityContext{
										Privileged:               proto.Bool(false),
										RunAsNonRoot:             proto.Bool(true),
										ReadOnlyRootFilesystem:   proto.Bool(true),
										AllowPrivilegeEscalation: proto.Bool(false),
										Capabilities: &corev1.Capabilities{
											Drop: []corev1.Capability{corev1.Capability("ALL")},
										},
									},
									ReadinessProbe: constants.GetRouterReadinessProbe(),
								},
							},
							Affinity:                     graph.Spec.Affinity,
							AutomountServiceAccountToken: proto.Bool(false), // Inference graph does not need access to api server
							Tolerations:                  graph.Spec.Tolerations,
							ImagePullSecrets:             config.GetImagePullSecrets(),
							NodeSelector:                 graph.Spec.NodeSelector,
							NodeName:                     graph.Spec.NodeName,
							ServiceAccountName:           graph.Spec.ServiceAccountName,
						},
					},
				},
			},
		},
	}

	// Only adding this env variable "PROPAGATE_HEADERS" if router's headers config has the key "propagate"
	value, exists := config.Headers["propagate"]
	if exists {
		service.Spec.ConfigurationSpec.Template.Spec.PodSpec.Containers[0].Env = []corev1.EnvVar{
			{
				Name:  constants.RouterHeadersPropagateEnvVar,
				Value: strings.Join(value, ","),
			},
		}
	}
	return service
}

func constructResourceRequirements(graph v1alpha1.InferenceGraph, config RouterConfig) corev1.ResourceRequirements {
	var specResources corev1.ResourceRequirements
	if !reflect.ValueOf(graph.Spec.Resources).IsZero() {
		log.Info("Ignoring defaults for ResourceRequirements as spec has resources mentioned", "specResources", graph.Spec.Resources)
		specResources = corev1.ResourceRequirements{
			Limits:   graph.Spec.Resources.Limits,
			Requests: graph.Spec.Resources.Requests,
		}
	} else {
		specResources = corev1.ResourceRequirements{
			Limits: corev1.ResourceList{
				corev1.ResourceCPU:    resource.MustParse(config.CpuLimit),
				corev1.ResourceMemory: resource.MustParse(config.MemoryLimit),
			},
			Requests: corev1.ResourceList{
				corev1.ResourceCPU:    resource.MustParse(config.CpuRequest),
				corev1.ResourceMemory: resource.MustParse(config.MemoryRequest),
			},
		}
	}
	return specResources
}