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 🙏

[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.

[ardaguclu]

Thank you for the steps and detailed descriptions about the issue. That really helped me to investigate.

As stated above, simply changing here with;

		r := o.Builder.
			Unstructured().
			Schema(o.Validator).
			Stream(o.In, "input").
			ContinueOnError().
			NamespaceParam(o.Namespace).DefaultNamespace().
			LabelSelectorParam(o.Selector).
			Flatten().
			Do()

resolves the problem (with additional note that cmd.Flags().Set("filename", "-") should be removed in unit test because it directly routes to FileVisitorForSTDIN which directly starts listening os.Stdin)

And I tried to find a elegant way for the users can easily use custom input instead of file or stdin. Unfortunately, there is none at least without introducing a new flag or a decorator function which may eventually cause more trouble in the future due to the incorrect usages.

I'm sorry to say this but I'd recommend using tmp file or directly stdin, I think we can close this issue as rejected.

[mfrancisc]

I tried to find a elegant way for the users can easily use custom input instead of file or stdin. Unfortunately, there is none at least without introducing a new flag or a decorator function which may eventually cause more trouble in the future due to the incorrect usages.

I see, @ardaguclu thanks for taking a look at it 🙏

I'm sorry to say this but I'd recommend using tmp file or directly stdin, I think we can close this issue as rejected.

Yeah , as I suspected, we discarded the tmp file option since it might turn out to be slow and extra work to be done for our controllers ( dumping ~100K objects per cluster to files/cleaning those up, and the number could grow ), as per the stdin option is not double for us - we would like to use this package from our controllers , thus it is not an interactive CLI application which can read from stdin.

[ardaguclu]

Although it may require several changes, this request looks reasonable.

In case anyone opens a PR

/triage accepted
/priority backlog

[tchap]

Can't you just set the objects in advance using SetObjects while loading the objects using a modified GetObjects?

Since this will cache the objects, the internal call to GetObjects will just return what you need.

Or am I missing something?

o.PreProcessorFn = func() error {
    infos, err := o.Builder.
        Unstructured().
        Schema(o.Validator).
        Stream(o.In, "input").
        ContinueOnError().
        NamespaceParam(o.Namespace).DefaultNamespace().
        LabelSelectorParam(o.Selector).
        Flatten().
        Do().
        Infos()
    if err != nil {
        return err
    }

    if o.ApplySet != nil {
        if err := o.ApplySet.AddLabels(infos...); err != nil {
            return err
        }
    }

    o.SetObjects(infos)
    return nil
}

Doesn't have to be in PreprocessorFn, I just put it there for fun.