Ability to apply objects from custom reader

[mfrancisc]

What would you like to be added:

Hello 👋 ,
I was trying to make use of the "k8s.io/kubectl/pkg/cmd/apply" package in order to apply unstructured.Unstructured objects we read from in memory or from CRs.

By doing that I needed to configure a custom reader ( thus not using stdin nor filesystem ), and I'm facing some issues. In other words, unless I've missed something this doesn't seem to be possible right now.

Following are my attempts:

1. Configure the apply command with custom reader using cobra command SetIn method:

package client

import (
	"io"
	"testing"

	"github.com/spf13/cobra"
	"github.com/stretchr/testify/require"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/cli-runtime/pkg/genericclioptions"
	"k8s.io/kubectl/pkg/cmd/apply"
	cmdtesting "k8s.io/kubectl/pkg/cmd/testing"
)

func TestCmdApply(t *testing.T) {
	// a random Unstructured object
	sa := &unstructured.Unstructured{
		Object: map[string]interface{}{
			"kind": "ServiceAccount",
			"metadata": map[string]interface{}{
				"name":      "test1",
				"namespace": "test",
			},
			"apiVersion": "v1",
		},
	}
	jsonContent, err := sa.MarshalJSON()
	require.NoError(t, err)
	// create a pipe with a reader and a writer for the above object
	r, w := io.Pipe()
	go func() {
		defer w.Close()
		w.Write(jsonContent)
	}()

	// configure apply cmd for testing
	ioStreams := genericclioptions.NewTestIOStreamsDiscard()
	f := cmdtesting.NewTestFactory()
	defer f.Cleanup()
	cmd := &cobra.Command{}
	flags := apply.NewApplyFlags(f, ioStreams)
	flags.AddFlags(cmd)
	cmd.Flags().Set("filename", "-")
	// configure apply cmd to read from the pipe
	cmd.SetIn(r)
	o, err := flags.ToOptions(cmd, "kubectl", []string{})
	if err != nil {
		t.Fatalf("unexpected error creating apply options: %s", err)
	}
	err = o.Validate(cmd, []string{})
	require.NoError(t, err)
	err = o.Run()
	require.NoError(t, err)
}

RESULT:

Received unexpected error: no objects passed to apply

2. Configure both the IOStreams and the cobra command with the custom reader:

package client

import (
   "io"
   "testing"

   "github.com/spf13/cobra"
   "github.com/stretchr/testify/require"
   "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
   "k8s.io/cli-runtime/pkg/genericclioptions"
   "k8s.io/kubectl/pkg/cmd/apply"
   cmdtesting "k8s.io/kubectl/pkg/cmd/testing"
)

func TestCmdApply(t *testing.T) {
   // a random Unstructured object
   sa := &unstructured.Unstructured{
   	Object: map[string]interface{}{
   		"kind": "ServiceAccount",
   		"metadata": map[string]interface{}{
   			"name":      "test1",
   			"namespace": "test",
   		},
   		"apiVersion": "v1",
   	},
   }
   jsonContent, err := sa.MarshalJSON()
   require.NoError(t, err)
   // create a pipe with a reader and a writer for the above object
   r, w := io.Pipe()
   go func() {
   	defer w.Close()
   	w.Write(jsonContent)
   }()

   // configure apply cmd for testing
   ioStreams := genericclioptions.NewTestIOStreamsDiscard()
   // set the reader from the pipe into the test streamer 
   ioStreams.In = r
   f := cmdtesting.NewTestFactory()
   defer f.Cleanup()
   cmd := &cobra.Command{}
   flags := apply.NewApplyFlags(f, ioStreams)
   flags.AddFlags(cmd)
   cmd.Flags().Set("filename", "-")
   // configure apply cmd to read from the pipe
   cmd.SetIn(r)
   o, err := flags.ToOptions(cmd, "kubectl", []string{})
   if err != nil {
   	t.Fatalf("unexpected error creating apply options: %s", err)
   }
   err = o.Validate(cmd, []string{})
   require.NoError(t, err)
   err = o.Run()
   require.NoError(t, err)
}

RESULT:

Received unexpected error: no objects passed to apply

3. Configure the resource.Builder with the customer reader from the pipe

package client

import (
	"io"
	"testing"

	"github.com/spf13/cobra"
	"github.com/stretchr/testify/require"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/cli-runtime/pkg/genericclioptions"
	"k8s.io/kubectl/pkg/cmd/apply"
	cmdtesting "k8s.io/kubectl/pkg/cmd/testing"
)

func TestCmdApply(t *testing.T) {
	// a random Unstructured object
	sa := &unstructured.Unstructured{
		Object: map[string]interface{}{
			"kind": "ServiceAccount",
			"metadata": map[string]interface{}{
				"name":      "test1",
				"namespace": "test",
			},
			"apiVersion": "v1",
		},
	}
	jsonContent, err := sa.MarshalJSON()
	require.NoError(t, err)
	// create a pipe with a reader and a writer for the above object
	r, w := io.Pipe()
	go func() {
		defer w.Close()
		w.Write(jsonContent)
	}()

	// configure apply cmd for testing
	ioStreams := genericclioptions.NewTestIOStreamsDiscard()
	ioStreams.In = r
	f := cmdtesting.NewTestFactory()
	defer f.Cleanup()
	cmd := &cobra.Command{}
	flags := apply.NewApplyFlags(f, ioStreams)
	flags.AddFlags(cmd)
	cmd.Flags().Set("filename", "-")
	// configure apply cmd to read from the pipe
	cmd.SetIn(r)
	o, err := flags.ToOptions(cmd, "kubectl", []string{})
	if err != nil {
		t.Fatalf("unexpected error creating apply options: %s", err)
	}
	o.Builder = o.Builder.Unstructured().Stream(r, "input")
	err = o.Validate(cmd, []string{})
	require.NoError(t, err)
	err = o.Run()
	require.NoError(t, err)
}

RESULT:

Received unexpected error: another mapper was already selected, cannot use unstructured types

4. Configure the resource.Builder with the customer reader from the pipe without the mapper:

package client

import (
"io"
"testing"

"github.com/spf13/cobra"
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/cli-runtime/pkg/genericclioptions"
"k8s.io/kubectl/pkg/cmd/apply"
cmdtesting "k8s.io/kubectl/pkg/cmd/testing"

)

func TestCmdApply(t *testing.T) {
// a random Unstructured object
sa := &unstructured.Unstructured{
Object: map[string]interface{}{
"kind": "ServiceAccount",
"metadata": map[string]interface{}{
"name": "test1",
"namespace": "test",
},
"apiVersion": "v1",
},
}
jsonContent, err := sa.MarshalJSON()
require.NoError(t, err)
// create a pipe with a reader and a writer for the above object
r, w := io.Pipe()
go func() {
defer w.Close()
w.Write(jsonContent)
}()

// configure apply cmd for testing
ioStreams := genericclioptions.NewTestIOStreamsDiscard()
ioStreams.In = r
f := cmdtesting.NewTestFactory()
defer f.Cleanup()
cmd := &cobra.Command{}
flags := apply.NewApplyFlags(f, ioStreams)
flags.AddFlags(cmd)
cmd.Flags().Set("filename", "-")
// configure apply cmd to read from the pipe
cmd.SetIn(r)
o, err := flags.ToOptions(cmd, "kubectl", []string{})
if err != nil {
	t.Fatalf("unexpected error creating apply options: %s", err)
}
o.Builder = o.Builder.Stream(r, "input")
err = o.Validate(cmd, []string{})
require.NoError(t, err)
err = o.Run()
require.NoError(t, err)

}

RESULT:

panic: runtime error: invalid memory address or nil pointer dereference [recovered]
	panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x2 addr=0x18 pc=0x1031acbb0]

goroutine 38 [running]:
...
k8s.io/cli-runtime/pkg/resource.(*mapper).infoForData(0x0, {0x1400089a000?, 0x1400088c3f0?, 0x0?}, {0x10365e06d, 0x5})
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/mapper.go:43 +0x30
k8s.io/cli-runtime/pkg/resource.(*StreamVisitor).Visit(0x14000364040, 0x140004c25e8)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:580 +0x150
k8s.io/cli-runtime/pkg/resource.EagerVisitorList.Visit({0x14000447200, 0x2, 0x103ca9f01?}, 0x14000364140)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:213 +0xf0
k8s.io/cli-runtime/pkg/resource.FlattenListVisitor.Visit({{0x103e340e0, 0x140004c21f8}, {0x103e3cc20, 0x104e361e8}, 0x1400088c300}, 0x14000364100)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:396 +0xbc
k8s.io/cli-runtime/pkg/resource.FlattenListVisitor.Visit({{0x103e34120, 0x1400088c330}, {0x103e3cc20, 0x104e361e8}, 0x1400088c300}, 0x140004c2378)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:396 +0xbc
k8s.io/cli-runtime/pkg/resource.ContinueOnErrorVisitor.Visit({{0x103e34120?, 0x1400088c390?}}, 0x140003640c0)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:359 +0xac
k8s.io/cli-runtime/pkg/resource.DecoratedVisitor.Visit({{0x103e340a0, 0x14000193220}, {0x14000447300, 0x3, 0x4}}, 0x14000193280)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/visitor.go:331 +0xc0
k8s.io/cli-runtime/pkg/resource.(*Result).Infos(0x140004a0580)
	/Users/fmuntean/go/pkg/mod/k8s.io/cli-runtime@v0.24.0/pkg/resource/result.go:122 +0xb0
k8s.io/kubectl/pkg/cmd/apply.(*ApplyOptions).GetObjects(0x14000582680)
	/Users/fmuntean/go/pkg/mod/k8s.io/kubectl@v0.24.0/pkg/cmd/apply/apply.go:407 +0x114
k8s.io/kubectl/pkg/cmd/apply.(*ApplyOptions).Run(0x14000582680)
...

In short: unless I've missed something, it doesn't seem to be possible to configure a custom reader with the current apply package implementation. Thanks in advance for your help.

Why is this needed:

We would love to be able to integrate the apply package into our components and be able to leverage features like 3WayMergePatch and ServerSide Apply when provisioning objects to kubernetes.
As anticipated we do not read those objects from files nor os.Stdin, instead we get those objects from other CRs, embed.FS , other sources that implement the io.Reader interface. And we would really like to avoid writing those objects to temporary files and read those from there , mainly because of performance issues and other constraints ( we are potentially handling thousands of objects and we need to provision those for the user in a timely manner ).

It would be nice if we could configure the resource.Builder upfront and skip the creation here which doesn't seem to be configurable with a custom Stream property, or any other way which could allow for really leveraging the stream based builder: o.Builder.Stream(r, "input") .

Any feedback/help is highly appreciated 🙏

[k8s-ci-robot]

This issue is currently awaiting triage.

SIG CLI takes a lead on issue triage for this repo, but any Kubernetes member can accept issues by applying the triage/accepted label.

The triage/accepted label can be added by org members by writing /triage accepted in a comment.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[ardaguclu]

I think, you can use stdin in IOStreams instead of initializing a dummy IOStream as in;
ioStreams := genericclioptions.NewTestIOStreamsDiscard()

[mfrancisc]

Hi @ardaguclu , thanks for your help.
Could you please elaborate more on this. How does this differ from point 2 in my attempts ?

[ardaguclu]

ah, sorry I didn't check point 2. Yes I was suggesting the same. If it doesn't work, this means this is not supported.