Repository: numaproj/numaflow (go)

Issue:
Default logs and exec to udf/udsink containers
# Summary

Make it simpler for first-time numaflow SDK users to look at their container logs or exec into their containers by defaulting `kubectl logs` and `kubectl exec` to use their container.  A few options:

1. Move the `udsink` and `udf` containers to be before the `main` container in the pod container list
2. Set `kubectl.kubernetes.io/default-container` pod annotation in controller
3. Keep things as-is, but have users set the annotation in the examples (at least ones in sdk repos) to look like:
```yaml
  - name: forward-message
    metadata:
      annotations:
        kubectl.kubernetes.io/default-container: udf
    udf:
      container:
         ...
```
```yaml
    - name: out
      metadata:
        annotations:
          kubectl.kubernetes.io/default-container: udsink
      sink:
        udsink:
          container:
            ...
```


# Use Cases

When developing a function or sink using one of the numaflow SDKs I'm regularly wanting to look at my container's logs.  When doing this I have to remember to add `-c udf` or `-c udsink` to kubectl commands or I get unexpected logs.

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request:
chore: default logs to user defined containers if there are. Close #283
Signed-off-by: Derek Wang <whynowy@gmail.com>

Close #283

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


Patch:
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index d50949ea..3fefa893 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -28,16 +28,17 @@ const (
 	Project = "numaflow"
 
 	// label/annotation keys.
-	KeyHash         = "numaflow.numaproj.io/hash" // hash of the object
-	KeyComponent    = "app.kubernetes.io/component"
-	KeyPartOf       = "app.kubernetes.io/part-of"
-	KeyManagedBy    = "app.kubernetes.io/managed-by"
-	KeyAppName      = "app.kubernetes.io/name"
-	KeyISBSvcName   = "numaflow.numaproj.io/isbsvc-name"
-	KeyISBSvcType   = "numaflow.numaproj.io/isbsvc-type"
-	KeyPipelineName = "numaflow.numaproj.io/pipeline-name"
-	KeyVertexName   = "numaflow.numaproj.io/vertex-name"
-	KeyReplica      = "numaflow.numaproj.io/replica"
+	KeyHash             = "numaflow.numaproj.io/hash" // hash of the object
+	KeyComponent        = "app.kubernetes.io/component"
+	KeyPartOf           = "app.kubernetes.io/part-of"
+	KeyManagedBy        = "app.kubernetes.io/managed-by"
+	KeyAppName          = "app.kubernetes.io/name"
+	KeyISBSvcName       = "numaflow.numaproj.io/isbsvc-name"
+	KeyISBSvcType       = "numaflow.numaproj.io/isbsvc-type"
+	KeyPipelineName     = "numaflow.numaproj.io/pipeline-name"
+	KeyVertexName       = "numaflow.numaproj.io/vertex-name"
+	KeyReplica          = "numaflow.numaproj.io/replica"
+	KeyDefaultContainer = "kubectl.kubernetes.io/default-container"
 
 	// ID key in the header of sources like http
 	KeyMetaID        = "x-numaflow-id"
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index 5faf0ac9..a1a2e74b 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -75,6 +75,10 @@ func (v Vertex) IsASink() bool {
 	return v.Spec.Sink != nil
 }
 
+func (v Vertex) IsUDSink() bool {
+	return v.IsASink() && v.Spec.Sink.UDSink != nil
+}
+
 func (v Vertex) IsMapUDF() bool {
 	return v.Spec.UDF != nil && v.Spec.UDF.GroupBy == nil
 }
diff --git a/pkg/reconciler/vertex/controller.go b/pkg/reconciler/vertex/controller.go
index 113a72ff..7664373e 100644
--- a/pkg/reconciler/vertex/controller.go
+++ b/pkg/reconciler/vertex/controller.go
@@ -215,6 +215,11 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 			labels[dfv1.KeyVertexName] = vertex.Spec.Name
 			annotations[dfv1.KeyHash] = hash
 			annotations[dfv1.KeyReplica] = strconv.Itoa(replica)
+			if vertex.IsMapUDF() || vertex.IsReduceUDF() {
+				annotations[dfv1.KeyDefaultContainer] = dfv1.CtrUdf
+			} else if vertex.IsUDSink() {
+				annotations[dfv1.KeyDefaultContainer] = dfv1.CtrUdsink
+			}
 			pod := &corev1.Pod{
 				ObjectMeta: metav1.ObjectMeta{
 					Namespace:       vertex.Namespace,


Test patch:
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
index 14734bf9..473e8b2c 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
@@ -355,6 +355,9 @@ func Test_VertexIsSink(t *testing.T) {
 	o := testVertex.DeepCopy()
 	o.Spec.Sink = &Sink{}
 	assert.True(t, o.IsASink())
+	assert.False(t, o.IsUDSink())
+	o.Spec.Sink.UDSink = &UDSink{}
+	assert.True(t, o.IsUDSink())
 }
 
 func Test_VertexGetInitContainers(t *testing.T) {