Skip to content

Commit

Permalink
[GEN-2267] chore: move limit device injection into webhook (#2307)
Browse files Browse the repository at this point in the history 
Co-authored-by: alonkeyval <[email protected]>
Co-authored-by: Amir Blum <[email protected]>
  • Loading branch information
@alonkeyval @alonbraymok @blumamir
3 people authored Feb 2, 2025
1 parent 4cf24cb commit 7fcfa95
Show file tree
Hide file tree
Showing 9 changed files with 311 additions and 399 deletions.
92 changes: 25 additions & 67 deletions instrumentor/controllers/instrumentationdevice/common.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (

"github.com/odigos-io/odigos/api/k8sconsts"
odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
"github.com/odigos-io/odigos/instrumentor/controllers/utils"
"github.com/odigos-io/odigos/instrumentor/controllers/utils/versionsupport"
"github.com/odigos-io/odigos/instrumentor/instrumentation"
"github.com/odigos-io/odigos/instrumentor/sdks"
Expand Down Expand Up @@ -64,52 +63,15 @@ func isDataCollectionReady(ctx context.Context, c client.Client) bool {
return nodeCollectorsGroup.Status.Ready
}

func addInstrumentationDeviceToWorkload(ctx context.Context, kubeClient client.Client, instConfig *odigosv1.InstrumentationConfig) (error, bool) {
// devicePartiallyApplied is used to indicate that the instrumentation device was partially applied for some of the containers.
devicePartiallyApplied := false
deviceNotAppliedDueToPresenceOfAnotherAgent := false
func enableOdigosInstrumentation(ctx context.Context, kubeClient client.Client, instConfig *odigosv1.InstrumentationConfig) error {

foundContainerWithSupportedLanguage := false
instrumentationSkippedDueToOtherAgent := false

logger := log.FromContext(ctx)
obj, err := getWorkloadObject(ctx, kubeClient, instConfig)
if err != nil {
return err, false
}

workload := k8sconsts.PodWorkload{
Name: obj.GetName(),
Namespace: obj.GetNamespace(),
Kind: k8sconsts.WorkloadKind(obj.GetObjectKind().GroupVersionKind().Kind),
}

// build an otel sdk map from instrumentation rules first, and merge it with the default otel sdk map
// this way, we can override the default otel sdk with the instrumentation rules
instrumentationRules := odigosv1.InstrumentationRuleList{}
err = kubeClient.List(ctx, &instrumentationRules)
if err != nil {
return err, false
}

// default otel sdk map according to Odigos tier
otelSdkToUse := GetDefaultSDKs()

for i := range instrumentationRules.Items {
instrumentationRule := &instrumentationRules.Items[i]
if instrumentationRule.Spec.Disabled || instrumentationRule.Spec.OtelSdks == nil {
// we only care about rules that have otel sdks configuration
continue
}

participating := utils.IsWorkloadParticipatingInRule(workload, instrumentationRule)
if !participating {
// filter rules that do not apply to the workload
continue
}

for lang, otelSdk := range instrumentationRule.Spec.OtelSdks.OtelSdkByLanguage {
// languages can override the default otel sdk or another rule.
// there is not check or warning if a language is defined in multiple rules at the moment.
otelSdkToUse[lang] = otelSdk
}
return err
}

result, err := controllerutil.CreateOrPatch(ctx, kubeClient, obj, func() error {
Expand All @@ -131,39 +93,34 @@ func addInstrumentationDeviceToWorkload(ctx context.Context, kubeClient client.C
agentsCanRunConcurrently = *odigosConfiguration.AllowConcurrentAgents
}

err, deviceApplied, deviceSkippedDueToOtherAgent := instrumentation.ApplyInstrumentationDevicesToPodTemplate(podSpec, instConfig.Status.RuntimeDetailsByContainer, otelSdkToUse, obj, logger, agentsCanRunConcurrently)
instrumentationSkippedDueToOtherAgent, foundContainerWithSupportedLanguage, err = instrumentation.ConfigureInstrumentationForPod(podSpec, instConfig.Status.RuntimeDetailsByContainer, obj, logger, agentsCanRunConcurrently)
if err != nil {
return err
}
// if non of the devices were applied due to the presence of another agent, return an error.
if !deviceApplied && deviceSkippedDueToOtherAgent {
deviceNotAppliedDueToPresenceOfAnotherAgent = true
}

devicePartiallyApplied = deviceSkippedDueToOtherAgent && deviceApplied
// If instrumentation device is applied successfully, add odigos.io/inject-instrumentation label to enable the webhook
if deviceApplied {
if !instrumentationSkippedDueToOtherAgent && foundContainerWithSupportedLanguage {
// add odigos.io/inject-instrumentation label to enable the webhook
instrumentation.SetInjectInstrumentationLabel(podSpec)
}

return nil

})

// if non of the devices were applied due to the presence of another agent, return an error.
if deviceNotAppliedDueToPresenceOfAnotherAgent {
return k8sutils.OtherAgentRunError, false
if instrumentationSkippedDueToOtherAgent {
return k8sutils.OtherAgentRunError
}

if err != nil {
return err, false
return err
}

modified := result != controllerutil.OperationResultNone
if modified {
logger.V(0).Info("added instrumentation device to workload", "name", obj.GetName(), "namespace", obj.GetNamespace())
logger.V(0).Info("inject instrumentation label to workload pod template", "name", obj.GetName(), "namespace", obj.GetNamespace())
}

return nil, devicePartiallyApplied
return nil
}

func removeInstrumentationDeviceFromWorkload(ctx context.Context, kubeClient client.Client, namespace string, workloadKind k8sconsts.WorkloadKind, workloadName string, uninstrumentReason ApplyInstrumentationDeviceReason) error {
Expand Down Expand Up @@ -278,17 +235,18 @@ func reconcileSingleWorkload(ctx context.Context, kubeClient client.Client, inst
return nil
}

err, devicePartiallyApplied := addInstrumentationDeviceToWorkload(ctx, kubeClient, instrumentationConfig)
if err == nil {
var successMessage string
if devicePartiallyApplied {
successMessage = "Instrumentation device partially applied"
} else {
successMessage = "Instrumentation device applied successfully"
}
conditions.UpdateStatusConditions(ctx, kubeClient, instrumentationConfig, &instrumentationConfig.Status.Conditions, metav1.ConditionTrue, appliedInstrumentationDeviceType, "InstrumentationDeviceApplied", successMessage)
err = enableOdigosInstrumentation(ctx, kubeClient, instrumentationConfig)
if err != nil {

conditions.UpdateStatusConditions(ctx, kubeClient, instrumentationConfig, &instrumentationConfig.Status.Conditions,
metav1.ConditionFalse, appliedInstrumentationDeviceType, string(ApplyInstrumentationDeviceReasonErrApplying),
"Odigos instrumentation failed to apply")
} else {
conditions.UpdateStatusConditions(ctx, kubeClient, instrumentationConfig, &instrumentationConfig.Status.Conditions, metav1.ConditionFalse, appliedInstrumentationDeviceType, string(ApplyInstrumentationDeviceReasonErrApplying), err.Error())

enabledMessage := "Odigos instrumentation is enabled"
conditions.UpdateStatusConditions(ctx, kubeClient, instrumentationConfig, &instrumentationConfig.Status.Conditions,
metav1.ConditionTrue, appliedInstrumentationDeviceType, "InstrumentationEnabled", enabledMessage)
}

return err
}
72 changes: 62 additions & 10 deletions instrumentor/controllers/instrumentationdevice/pods_webhook.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,20 +7,22 @@ import (

"github.com/go-logr/logr"
"github.com/odigos-io/odigos/api/k8sconsts"
odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
"github.com/odigos-io/odigos/common"
"github.com/odigos-io/odigos/instrumentor/controllers/utils"
webhookdeviceinjector "github.com/odigos-io/odigos/instrumentor/internal/webhook_device_injector"
webhookenvinjector "github.com/odigos-io/odigos/instrumentor/internal/webhook_env_injector"
containerutils "github.com/odigos-io/odigos/k8sutils/pkg/container"
"github.com/odigos-io/odigos/instrumentor/sdks"
sourceutils "github.com/odigos-io/odigos/k8sutils/pkg/source"
"github.com/odigos-io/odigos/k8sutils/pkg/workload"
"go.opentelemetry.io/otel/attribute"
semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/controller-runtime/pkg/webhook"
"sigs.k8s.io/controller-runtime/pkg/webhook/admission"

corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
)

const otelServiceNameEnvVarName = "OTEL_SERVICE_NAME"
Expand Down Expand Up @@ -49,11 +51,11 @@ func (p *PodsWebhook) Default(ctx context.Context, obj runtime.Object) error {
pod.Annotations = map[string]string{}
}

// Inject ODIGOS environment variables into all containers
return p.injectOdigosEnvVars(ctx, logger, pod)
// Inject ODIGOS environment variables and instrumentation device into all containers
return p.injectOdigosInstrumentation(ctx, logger, pod)
}

func (p *PodsWebhook) injectOdigosEnvVars(ctx context.Context, logger logr.Logger, pod *corev1.Pod) error {
func (p *PodsWebhook) injectOdigosInstrumentation(ctx context.Context, logger logr.Logger, pod *corev1.Pod) error {
// Environment variables that remain consistent across all containers
commonEnvVars := getCommonEnvVars()

Expand All @@ -79,19 +81,39 @@ func (p *PodsWebhook) injectOdigosEnvVars(ctx context.Context, logger logr.Logge
return fmt.Errorf("namespace is empty for pod %s/%s, Skipping Injection of ODIGOS environment variables", pod.Namespace, pod.Name)
}
}
var workloadInstrumentationConfig odigosv1.InstrumentationConfig
instrumentationConfigName := workload.CalculateWorkloadRuntimeObjectName(podWorkload.Name, podWorkload.Kind)

if err := p.Get(ctx, client.ObjectKey{Namespace: podWorkload.Namespace, Name: instrumentationConfigName}, &workloadInstrumentationConfig); err != nil {
return fmt.Errorf("failed to get instrumentationConfig: %w", err)
}

otelSdkToUse, err := getRelevantOtelSDKs(ctx, p.Client, *podWorkload)
if err != nil {
return fmt.Errorf("failed to determine OpenTelemetry SDKs: %w", err)
}

var serviceName *string
var serviceNameEnv *corev1.EnvVar

for i := range pod.Spec.Containers {
container := &pod.Spec.Containers[i]
runtimeDetails := workloadInstrumentationConfig.Status.GetRuntimeDetailsForContainer(*container)
if runtimeDetails == nil {
continue
}

if runtimeDetails.Language == common.UnknownProgrammingLanguage {
continue
}

pl, otelsdk, found := containerutils.GetLanguageAndOtelSdk(container)
otelSdk, found := otelSdkToUse[runtimeDetails.Language]
if !found {
continue
}

webhookenvinjector.InjectOdigosAgentEnvVars(ctx, p.Client, logger, *podWorkload, container, pl, otelsdk)
webhookdeviceinjector.InjectOdigosInstrumentationDevice(*podWorkload, container, otelSdk, runtimeDetails)
webhookenvinjector.InjectOdigosAgentEnvVars(logger, *podWorkload, container, otelSdk, runtimeDetails)

// Check if the environment variables are already present, if so skip inject them again.
if envVarsExist(container.Env, commonEnvVars) {
Expand All @@ -101,7 +123,7 @@ func (p *PodsWebhook) injectOdigosEnvVars(ctx context.Context, logger logr.Logge
containerNameEnv := corev1.EnvVar{Name: k8sconsts.OdigosEnvVarContainerName, Value: container.Name}
container.Env = append(container.Env, append(commonEnvVars, containerNameEnv)...)

if shouldInjectServiceName(pl, otelsdk) {
if shouldInjectServiceName(runtimeDetails.Language, otelSdk) {
// Ensure the serviceName is fetched only once per pod
if serviceName == nil {
serviceName = p.getServiceNameForEnv(ctx, logger, podWorkload)
Expand Down Expand Up @@ -251,3 +273,33 @@ func (p *PodsWebhook) getServiceNameForEnv(ctx context.Context, logger logr.Logg

return &resolvedServiceName
}

func getRelevantOtelSDKs(ctx context.Context, kubeClient client.Client, podWorkload k8sconsts.PodWorkload) (map[common.ProgrammingLanguage]common.OtelSdk, error) {

instrumentationRules := odigosv1.InstrumentationRuleList{}
if err := kubeClient.List(ctx, &instrumentationRules); err != nil {
return nil, err
}

otelSdkToUse := sdks.GetDefaultSDKs()
for i := range instrumentationRules.Items {
rule := &instrumentationRules.Items[i]
if rule.Spec.Disabled || rule.Spec.OtelSdks == nil {
// we only care about rules that have otel sdks configuration
continue
}

if !utils.IsWorkloadParticipatingInRule(podWorkload, rule) {
// filter rules that do not apply to the workload
continue
}

for lang, otelSdk := range rule.Spec.OtelSdks.OtelSdkByLanguage {
// languages can override the default otel sdk or another rule.
// there is not check or warning if a language is defined in multiple rules at the moment.
otelSdkToUse[lang] = otelSdk
}
}

return otelSdkToUse, nil
}
43 changes: 9 additions & 34 deletions instrumentor/instrumentation/instrumentation.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package instrumentation

import (
"errors"
"fmt"
"strings"

"github.com/go-logr/logr"
Expand All @@ -12,37 +11,35 @@ import (

"github.com/odigos-io/odigos/common"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
)

var (
ErrNoDefaultSDK = errors.New("no default sdks found")
)

func ApplyInstrumentationDevicesToPodTemplate(original *corev1.PodTemplateSpec, runtimeDetails []odigosv1.RuntimeDetailsByContainer, defaultSdks map[common.ProgrammingLanguage]common.OtelSdk, targetObj client.Object,
logger logr.Logger, agentsCanRunConcurrently bool) (error, bool, bool) {
func ConfigureInstrumentationForPod(original *corev1.PodTemplateSpec, runtimeDetails []odigosv1.RuntimeDetailsByContainer, targetObj client.Object,
logger logr.Logger, agentsCanRunConcurrently bool) (bool, bool, error) {
// delete any existing instrumentation devices.
// this is necessary for example when migrating from community to enterprise,
// and we need to cleanup the community device before adding the enterprise one.
RevertInstrumentationDevices(original)

deviceApplied := false
deviceSkippedDueToOtherAgent := false
foundContainerWithSupportedLanguage := false
instrumentationSkippedDueToOtherAgent := false
var modifiedContainers []corev1.Container

for _, container := range original.Spec.Containers {
containerLanguage := getLanguageOfContainer(runtimeDetails, container.Name)
containerHaveOtherAgent := getContainerOtherAgents(runtimeDetails, container.Name)
libcType := getLibCTypeOfContainer(runtimeDetails, container.Name)

// By default, Odigos does not run alongside other agents.
// However, if configured in the odigos-config, it can be allowed to run in parallel.
if containerHaveOtherAgent != nil && !agentsCanRunConcurrently {
logger.Info("Container is running other agent, skip applying instrumentation device", "agent", containerHaveOtherAgent.Name, "container", container.Name)
logger.Info("Container is running other agent, skip applying instrumentation label", "agent", containerHaveOtherAgent.Name, "container", container.Name)

// Not actually modifying the container, but we need to append it to the list.
modifiedContainers = append(modifiedContainers, container)
deviceSkippedDueToOtherAgent = true
instrumentationSkippedDueToOtherAgent = true
continue
}

Expand All @@ -53,30 +50,18 @@ func ApplyInstrumentationDevicesToPodTemplate(original *corev1.PodTemplateSpec,
// TODO: this will make it look as if instrumentation device is applied,
// which is incorrect
modifiedContainers = append(modifiedContainers, container)
continue
}

// Find and apply the appropriate SDK for the container language.
otelSdk, found := defaultSdks[containerLanguage]
if !found {
return fmt.Errorf("%w for language: %s, container:%s", ErrNoDefaultSDK, containerLanguage, container.Name), deviceApplied, deviceSkippedDueToOtherAgent
}

instrumentationDeviceName := common.InstrumentationDeviceName(containerLanguage, otelSdk, libcType)
if container.Resources.Limits == nil {
container.Resources.Limits = make(map[corev1.ResourceName]resource.Quantity)
continue
}
container.Resources.Limits[corev1.ResourceName(instrumentationDeviceName)] = resource.MustParse("1")
deviceApplied = true

foundContainerWithSupportedLanguage = true
modifiedContainers = append(modifiedContainers, container)
}

if modifiedContainers != nil {
original.Spec.Containers = modifiedContainers
}

return nil, deviceApplied, deviceSkippedDueToOtherAgent
return instrumentationSkippedDueToOtherAgent, foundContainerWithSupportedLanguage, nil
}

func RevertInstrumentationDevices(original *corev1.PodTemplateSpec) bool {
Expand Down Expand Up @@ -120,16 +105,6 @@ func getContainerOtherAgents(runtimeDetails []odigosv1.RuntimeDetailsByContainer
return nil
}

func getLibCTypeOfContainer(runtimeDetails []odigosv1.RuntimeDetailsByContainer, containerName string) *common.LibCType {
for _, rd := range runtimeDetails {
if rd.ContainerName == containerName {
return rd.LibCType
}
}

return nil
}

func SetInjectInstrumentationLabel(original *corev1.PodTemplateSpec) {

if original.Labels == nil {
Expand Down
Loading

0 comments on commit 7fcfa95

Please sign in to comment.