Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: odigos describe when pipeline is not created #1622

Merged
merged 3 commits into from
Oct 26, 2024
Merged

fix: odigos describe when pipeline is not created #1622

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
57fde68
fix: describe text when odigos pipeline is not yet deployed
blumamir Oct 26, 2024
dcd8764
fix: expecting cluster collector when there are only destinations
blumamir Oct 26, 2024
efae431
feat: add status for collectors
blumamir Oct 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions .vscode/launch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,16 @@
"cwd": "${workspaceFolder}/cli",
"args": ["uninstall", "--yes"],
"buildFlags": "-tags=embed_manifests"
},
{
"name": "cli describe",
"type": "go",
"request": "launch",
"mode": "debug",
"program": "${workspaceFolder}/cli",
"cwd": "${workspaceFolder}/cli",
"args": ["describe"],
"buildFlags": "-tags=embed_manifests"
}
]
}
232 changes: 138 additions & 94 deletions k8sutils/pkg/describe/odigos.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,23 +34,27 @@ type odigosResources struct {
InstrumentationConfigs *odigosv1.InstrumentationConfigList
}

func getClusterCollectorResources(ctx context.Context, kubeClient kubernetes.Interface, odigosClient odigosclientset.OdigosV1alpha1Interface, odigosNs string) (clusterCollector clusterCollectorResources, err error) {
func getClusterCollectorResources(ctx context.Context, kubeClient kubernetes.Interface, odigosClient odigosclientset.OdigosV1alpha1Interface, odigosNs string) (clusterCollectorResources, error) {

clusterCollector = clusterCollectorResources{}
clusterCollector := clusterCollectorResources{}

clusterCollector.CollectorsGroup, err = odigosClient.CollectorsGroups(odigosNs).Get(ctx, consts.OdigosClusterCollectorCollectorGroupName, metav1.GetOptions{})
if err != nil && !apierrors.IsNotFound(err) {
return
cg, err := odigosClient.CollectorsGroups(odigosNs).Get(ctx, consts.OdigosClusterCollectorCollectorGroupName, metav1.GetOptions{})
if err == nil {
clusterCollector.CollectorsGroup = cg
} else if !apierrors.IsNotFound(err) {
return clusterCollectorResources{}, err
}

clusterCollector.Deployment, err = kubeClient.AppsV1().Deployments(odigosNs).Get(ctx, consts.OdigosClusterCollectorDeploymentName, metav1.GetOptions{})
if err != nil && !apierrors.IsNotFound(err) {
return
dep, err := kubeClient.AppsV1().Deployments(odigosNs).Get(ctx, consts.OdigosClusterCollectorDeploymentName, metav1.GetOptions{})
if err == nil {
clusterCollector.Deployment = dep
} else if !apierrors.IsNotFound(err) {
return clusterCollectorResources{}, err
}

var clusterRoleRevision string
if clusterCollector.Deployment != nil {
revisionAnnotation, found := clusterCollector.Deployment.Annotations["deployment.kubernetes.io/revision"]
if dep != nil {
revisionAnnotation, found := dep.Annotations["deployment.kubernetes.io/revision"]
if found {
clusterRoleRevision = revisionAnnotation
}
Expand All @@ -62,8 +66,7 @@ func getClusterCollectorResources(ctx context.Context, kubeClient kubernetes.Int
LabelSelector: metav1.FormatLabelSelector(clusterCollector.Deployment.Spec.Selector),
})
if errRs != nil {
err = errRs
return
return clusterCollectorResources{}, errRs
}

var latestRevisionReplicaSet *appsv1.ReplicaSet
Expand All @@ -81,29 +84,33 @@ func getClusterCollectorResources(ctx context.Context, kubeClient kubernetes.Int
LabelSelector: fmt.Sprintf("pod-template-hash=%s", podTemplateHash),
})
if err != nil {
return
return clusterCollectorResources{}, err
}
}
}

return
return clusterCollector, nil
}

func getNodeCollectorResources(ctx context.Context, kubeClient kubernetes.Interface, odigosClient odigosclientset.OdigosV1alpha1Interface, odigosNs string) (nodeCollector nodeCollectorResources, err error) {
func getNodeCollectorResources(ctx context.Context, kubeClient kubernetes.Interface, odigosClient odigosclientset.OdigosV1alpha1Interface, odigosNs string) (nodeCollectorResources, error) {

nodeCollector = nodeCollectorResources{}
nodeCollector := nodeCollectorResources{}

nodeCollector.CollectorsGroup, err = odigosClient.CollectorsGroups(odigosNs).Get(ctx, consts.OdigosNodeCollectorCollectorGroupName, metav1.GetOptions{})
if err != nil && !apierrors.IsNotFound(err) {
return
cg, err := odigosClient.CollectorsGroups(odigosNs).Get(ctx, consts.OdigosNodeCollectorCollectorGroupName, metav1.GetOptions{})
if err == nil {
nodeCollector.CollectorsGroup = cg
} else if !apierrors.IsNotFound(err) {
return nodeCollectorResources{}, err
}

nodeCollector.DaemonSet, err = kubeClient.AppsV1().DaemonSets(odigosNs).Get(ctx, consts.OdigosNodeCollectorDaemonSetName, metav1.GetOptions{})
if err != nil && !apierrors.IsNotFound(err) {
return
ds, err := kubeClient.AppsV1().DaemonSets(odigosNs).Get(ctx, consts.OdigosNodeCollectorDaemonSetName, metav1.GetOptions{})
if err == nil {
nodeCollector.DaemonSet = ds
} else if !apierrors.IsNotFound(err) {
return nodeCollectorResources{}, err
}

return
return nodeCollector, nil
}

func getRelevantOdigosResources(ctx context.Context, kubeClient kubernetes.Interface, odigosClient odigosclientset.OdigosV1alpha1Interface, odigosNs string) (odigos odigosResources, err error) {
Expand Down Expand Up @@ -135,47 +142,56 @@ func printOdigosVersion(odigosVersion string, sb *strings.Builder) {
describeText(sb, 0, "Odigos Version: %s", odigosVersion)
}

func printOdigosPipelineStatus(numInstrumentationConfigs, numDestinations int, expectingPipeline bool, sb *strings.Builder) {
if expectingPipeline {
describeText(sb, 1, "Status: there are %d sources and %d destinations so pipeline will be deployed\n", numInstrumentationConfigs, numDestinations)
} else {
describeText(sb, 1, "Status: no sources or destinations found so pipeline will not be deployed")
}
}
func printClusterCollectorStatus(odigosResources odigosResources, destinations *odigosv1.DestinationList, sb *strings.Builder) {

expectingClusterCollector := len(destinations.Items) > 0

func printClusterCollectorStatus(clusterCollector clusterCollectorResources, expectingPipeline bool, sb *strings.Builder) {
describeText(sb, 1, "Cluster Collector:")
if clusterCollector.CollectorsGroup == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Not Created", !expectingPipeline))
return
clusterCollector := odigosResources.ClusterCollector

if expectingClusterCollector {
describeText(sb, 2, "Status: Cluster Collector is expected to be created because there are destinations")
} else {
describeText(sb, 2, "Status: Cluster Collector is not expected to be created because there are no destinations")
}

describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Created", expectingPipeline))
if clusterCollector.CollectorsGroup == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Not Created", !expectingClusterCollector))
} else {
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Created", expectingClusterCollector))

var deployedCondition *metav1.Condition
for _, condition := range clusterCollector.CollectorsGroup.Status.Conditions {
if condition.Type == "Deployed" {
deployedCondition = &condition
break
var deployedCondition *metav1.Condition
for _, condition := range clusterCollector.CollectorsGroup.Status.Conditions {
if condition.Type == "Deployed" {
deployedCondition = &condition
break
}
}
}
if deployedCondition == nil {
describeText(sb, 2, wrapTextInRed("Deployed: Status Unavailable"))
} else {
if deployedCondition.Status == metav1.ConditionTrue {
describeText(sb, 2, wrapTextInGreen("Deployed: true"))
if deployedCondition == nil {
describeText(sb, 2, wrapTextInRed("Deployed: Status Unavailable"))
} else {
describeText(sb, 2, wrapTextInRed("Deployed: false"))
describeText(sb, 2, wrapTextInRed(fmt.Sprintf("Reason: %s", deployedCondition.Message)))
if deployedCondition.Status == metav1.ConditionTrue {
describeText(sb, 2, wrapTextInGreen("Deployed: true"))
} else {
describeText(sb, 2, wrapTextInRed("Deployed: false"))
describeText(sb, 2, wrapTextInRed(fmt.Sprintf("Reason: %s", deployedCondition.Message)))
}
}
}

ready := clusterCollector.CollectorsGroup.Status.Ready
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Ready: %t", ready), ready))
ready := clusterCollector.CollectorsGroup.Status.Ready
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Ready: %t", ready), ready))
}

if clusterCollector.LatestRevisionPods == nil || clusterCollector.Deployment == nil {
describeText(sb, 2, wrapTextInRed("Number of Replicas: Status Unavailable"))
expectedReplicas := int32(0)
if clusterCollector.Deployment == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("Deployment: Not Found", !expectingClusterCollector))
} else {
describeText(sb, 2, wrapTextSuccessOfFailure("Deployment: Found", expectingClusterCollector))
expectedReplicas = *clusterCollector.Deployment.Spec.Replicas
describeText(sb, 2, fmt.Sprintf("Expected Replicas: %d", expectedReplicas))
}

if clusterCollector.LatestRevisionPods != nil {
runningReplicas := 0
failureReplicas := 0
var failureText string
Expand All @@ -199,72 +215,100 @@ func printClusterCollectorStatus(clusterCollector clusterCollectorResources, exp
}
}
}
desiredReplicas := *clusterCollector.Deployment.Spec.Replicas
describeText(sb, 2, fmt.Sprintf("Desired Replicas: %d", desiredReplicas))
podReplicasText := fmt.Sprintf("Actual Replicas: %d running, %d failed", runningReplicas, failureReplicas)
deploymentSuccessful := runningReplicas == int(desiredReplicas)
deploymentSuccessful := runningReplicas == int(expectedReplicas) && failureReplicas == 0
describeText(sb, 2, wrapTextSuccessOfFailure(podReplicasText, deploymentSuccessful))
if !deploymentSuccessful {
describeText(sb, 2, wrapTextInRed(fmt.Sprintf("Replicas Not Ready Reason: %s", failureText)))
}
}

}

func printNodeCollectorStatus(nodeCollector nodeCollectorResources, expectingNodeCollector bool, sb *strings.Builder) {
describeText(sb, 1, "Node Collector:")
if nodeCollector.CollectorsGroup == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Not Created", !expectingNodeCollector))
return
func printAndCalculateIsNodeCollectorStatus(odigosResources *odigosResources, sb *strings.Builder) bool {

numInstrumentationConfigs := len(odigosResources.InstrumentationConfigs.Items)
if numInstrumentationConfigs == 0 {
describeText(sb, 2, "Status: Node Collectors not expected as there are no sources")
return false
}

describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Created", expectingNodeCollector))
if odigosResources.ClusterCollector.CollectorsGroup == nil {
describeText(sb, 2, "Status: Node Collectors not expected as there are no destinations")
return false
}

var deployedCondition *metav1.Condition
for _, condition := range nodeCollector.CollectorsGroup.Status.Conditions {
if condition.Type == "Deployed" {
deployedCondition = &condition
break
}
if !odigosResources.ClusterCollector.CollectorsGroup.Status.Ready {
describeText(sb, 2, "Status: Node Collectors not expected as the Cluster Collector is not ready")
return false
}
if deployedCondition == nil {
describeText(sb, 2, wrapTextInRed("Deployed: Status Unavailable"))

describeText(sb, 2, "Status: Node Collectors expected as cluster collector is ready and there are sources")
return true
}

func printNodeCollectorStatus(odigosResources odigosResources, sb *strings.Builder) {

describeText(sb, 1, "Node Collector:")
nodeCollector := odigosResources.NodeCollector

expectingNodeCollector := printAndCalculateIsNodeCollectorStatus(&odigosResources, sb)

if nodeCollector.CollectorsGroup == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Not Created", !expectingNodeCollector))
} else {
if deployedCondition.Status == metav1.ConditionTrue {
describeText(sb, 2, wrapTextInGreen("Deployed: True"))
describeText(sb, 2, wrapTextSuccessOfFailure("Collectors Group Created", expectingNodeCollector))

var deployedCondition *metav1.Condition
for _, condition := range nodeCollector.CollectorsGroup.Status.Conditions {
if condition.Type == "Deployed" {
deployedCondition = &condition
break
}
}
if deployedCondition == nil {
describeText(sb, 2, wrapTextInRed("Deployed: Status Unavailable"))
} else {
describeText(sb, 2, wrapTextInRed("Deployed: False"))
describeText(sb, 2, wrapTextInRed(fmt.Sprintf("Reason: %s", deployedCondition.Message)))
if deployedCondition.Status == metav1.ConditionTrue {
describeText(sb, 2, wrapTextInGreen("Deployed: True"))
} else {
describeText(sb, 2, wrapTextInRed("Deployed: False"))
describeText(sb, 2, wrapTextInRed(fmt.Sprintf("Reason: %s", deployedCondition.Message)))
}
}

ready := nodeCollector.CollectorsGroup.Status.Ready
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Ready: %t", ready), ready))
}

ready := nodeCollector.CollectorsGroup.Status.Ready
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Ready: %t", ready), ready))

// this is copied from k8sutils/pkg/describe/describe.go
// I hope the info is accurate since there can be many edge cases
describeText(sb, 2, "Desired Number of Nodes Scheduled: %d", nodeCollector.DaemonSet.Status.DesiredNumberScheduled)
currentMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.CurrentNumberScheduled
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Current Number of Nodes Scheduled: %d", nodeCollector.DaemonSet.Status.CurrentNumberScheduled), currentMeetsDesired))
updatedMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.UpdatedNumberScheduled
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Scheduled with Up-to-date Pods: %d", nodeCollector.DaemonSet.Status.UpdatedNumberScheduled), updatedMeetsDesired))
availableMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.NumberAvailable
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Scheduled with Available Pods: %d", nodeCollector.DaemonSet.Status.NumberAvailable), availableMeetsDesired))
noMisscheduled := nodeCollector.DaemonSet.Status.NumberMisscheduled == 0
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Misscheduled: %d", nodeCollector.DaemonSet.Status.NumberMisscheduled), noMisscheduled))
if nodeCollector.DaemonSet == nil {
describeText(sb, 2, wrapTextSuccessOfFailure("DaemonSet: Not Found", !expectingNodeCollector))
} else {
describeText(sb, 2, wrapTextSuccessOfFailure("DaemonSet: Found", expectingNodeCollector))

// this is copied from k8sutils/pkg/describe/describe.go
// I hope the info is accurate since there can be many edge cases
describeText(sb, 2, "Desired Number of Nodes Scheduled: %d", nodeCollector.DaemonSet.Status.DesiredNumberScheduled)
currentMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.CurrentNumberScheduled
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Current Number of Nodes Scheduled: %d", nodeCollector.DaemonSet.Status.CurrentNumberScheduled), currentMeetsDesired))
updatedMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.UpdatedNumberScheduled
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Scheduled with Up-to-date Pods: %d", nodeCollector.DaemonSet.Status.UpdatedNumberScheduled), updatedMeetsDesired))
availableMeetsDesired := nodeCollector.DaemonSet.Status.DesiredNumberScheduled == nodeCollector.DaemonSet.Status.NumberAvailable
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Scheduled with Available Pods: %d", nodeCollector.DaemonSet.Status.NumberAvailable), availableMeetsDesired))
noMisscheduled := nodeCollector.DaemonSet.Status.NumberMisscheduled == 0
describeText(sb, 2, wrapTextSuccessOfFailure(fmt.Sprintf("Number of Nodes Misscheduled: %d", nodeCollector.DaemonSet.Status.NumberMisscheduled), noMisscheduled))
}
}

func printOdigosPipeline(odigosResources odigosResources, sb *strings.Builder) {
describeText(sb, 0, "Odigos Pipeline:")
numDestinations := len(odigosResources.Destinations.Items)
numInstrumentationConfigs := len(odigosResources.InstrumentationConfigs.Items)
// odigos will only initiate pipeline if there are any sources or destinations
expectingPipeline := numDestinations > 0 || numInstrumentationConfigs > 0

printOdigosPipelineStatus(numInstrumentationConfigs, numDestinations, expectingPipeline, sb)
printClusterCollectorStatus(odigosResources.ClusterCollector, expectingPipeline, sb)
describeText(sb, 1, "Status: there are %d sources and %d destinations\n", numInstrumentationConfigs, numDestinations)
printClusterCollectorStatus(odigosResources, odigosResources.Destinations, sb)
sb.WriteString("\n")
expectingNodeCollector := odigosResources.ClusterCollector.CollectorsGroup != nil && odigosResources.ClusterCollector.CollectorsGroup.Status.Ready && numInstrumentationConfigs > 0
printNodeCollectorStatus(odigosResources.NodeCollector, expectingNodeCollector, sb)
printNodeCollectorStatus(odigosResources, sb)
}

func printDescribeOdigos(odigosVersion string, odigosResources odigosResources) string {
Expand Down
Loading