Example for create object from file via dynamic API

[yuzhichang]

I need code to create object from file just like kubectl since I don't want to depend on kubectl executable.

The following code works for creating Deployment, Service etc but not work for ClusterRole, ClusterRoleBind etc. The error is:
client.Resource.Namespace.Create failed: the server could not find the requested resource.

Could you help to fix it?

package main

import (
	"context"
	"flag"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"sort"

	"github.com/thanos-io/thanos/pkg/errors"
	apiv1 "k8s.io/api/core/v1"
	apiErrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
	"sigs.k8s.io/yaml"
)

func kind2Resource(group, version, kind string) (gvr schema.GroupVersionResource, err error) {
	// https://www.cnblogs.com/zhangmingcheng/p/16128224.html
	// https://iximiuz.com/en/posts/kubernetes-api-structure-and-terminology/
	// https://github.com/kubernetes/kubernetes/issues/18622
	// kubectl api-resources
	gvk := schema.GroupVersionKind{Group: group, Version: version, Kind: kind}
	gvr, _ = meta.UnsafeGuessKindToResource(gvk)
	//fmt.Printf("%s %s %s -> %+v\n", group, version, kind, gvr)
	return
}

func createObject(client dynamic.Interface, fp string) (err error) {
	fmt.Printf("Reading %s...\n", fp)
	var b []byte
	if b, err = os.ReadFile(fp); err != nil {
		err = errors.Wrapf(err, "os.ReadFile failed")
		return
	}
	m := make(map[string]interface{})
	if err = yaml.Unmarshal(b, &m); err != nil {
		err = errors.Wrapf(err, "yaml.Unmarshal failed")
		return
	}

	obj := unstructured.Unstructured{Object: m}
	var apiVersion, kind, namespace string
	apiVersion = obj.GetAPIVersion()
	kind = obj.GetKind()
	namespace = obj.GetNamespace()
	if namespace == "" {
		namespace = apiv1.NamespaceDefault
	}
	gv, _ := schema.ParseGroupVersion(apiVersion)
	var gvr schema.GroupVersionResource
	if gvr, err = kind2Resource(gv.Group, gv.Version, kind); err != nil {
		return
	}
	var result *unstructured.Unstructured
	result, err = client.Resource(gvr).Namespace(namespace).Create(context.TODO(), &obj, metav1.CreateOptions{})
	if err != nil {
		err = errors.Wrapf(err, "client.Resource.Namespace.Create failed")
		return
	}
	fmt.Printf("Created resource %q.\n", result.GetName())
	return
}

func main() {
	// Initialize client
	var kubeconfig *string
	if home := homedir.HomeDir(); home != "" {
		kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
	} else {
		kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
	}
	pth := flag.String("f", ".", "path of manifest file or directory")
	flag.Parse()

	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		panic(err)
	}
	client, err := dynamic.NewForConfig(config)
	if err != nil {
		panic(err)
	}

	// Get manifest list
	var fi fs.FileInfo
	if fi, err = os.Stat(*pth); err != nil {
		panic(err)
	}
	var fps []string
	if fi.IsDir() {
		var des []fs.DirEntry
		if des, err = os.ReadDir(*pth); err != nil {
			panic(err)
		}
		for _, de := range des {
			if de.IsDir() {
				continue
			}
			fn := de.Name()
			ext := filepath.Ext(fn)
			if ext != ".yaml" && ext != "yml" {
				continue
			}
			fps = append(fps, filepath.Join(*pth, fn))
		}
		sort.Strings(fps)
	} else {
		fps = append(fps, *pth)
	}

	// Create resource for each manifest
	for _, fp := range fps {
		if err = createObject(client, fp); err != nil {
			if !apiErrors.IsAlreadyExists(err) {
				fmt.Println(err)
			}
		}
	}
}

# blackboxExporter-clusterRoleBinding.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  labels:
    app.kubernetes.io/component: exporter
    app.kubernetes.io/name: blackbox-exporter
    app.kubernetes.io/part-of: kube-prometheus
    app.kubernetes.io/version: 0.22.0
  name: blackbox-exporter
  namespace: monitoring
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: blackbox-exporter
subjects:
- kind: ServiceAccount
  name: blackbox-exporter
  namespace: monitoring

[iximiuz]

There might be an alternative solution - check out the "Read Kubernetes Manifests Into Go Structs" section of this article of mine. And here is the corresponding mini-program from this repo. Hope this helps!

[yuzhichang]

@iximiuz info.Object has different type with unstructured.Unstructured.Object. I cannot figure out how to use info.Object with dynamic API. Could you provide a complete example?

[iximiuz]

@yuzhichang info.Object is of the type runtime.Object. It's a special interface type that can front many concrete API types, in particular unstructured.Unstructured.

Here is an example of how to type-cast a runtime.Object into a concrete unstructured object. And you can read more about the Kubernetes API types in another my article.

[yuzhichang]

I've figured out the reason of error client.Resource.Namespace.Create failed: the server could not find the requested resource.
Cluster-wide resources' namespace shall not be specified at creation.

The complete code is the following:

package main

import (
	"context"
	"flag"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"github.com/thanos-io/thanos/pkg/errors"
	apiv1 "k8s.io/api/core/v1"
	apiErrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/yaml"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func kind2Resource(group, version, kind string) (gvr schema.GroupVersionResource, err error) {
	// https://www.cnblogs.com/zhangmingcheng/p/16128224.html
	// https://iximiuz.com/en/posts/kubernetes-api-structure-and-terminology/
	// https://github.com/kubernetes/kubernetes/issues/18622
	// kubectl api-resources
	gvk := schema.GroupVersionKind{Group: group, Version: version, Kind: kind}
	gvr, _ = meta.UnsafeGuessKindToResource(gvk)
	//fmt.Printf("%s %s %s -> %+v\n", group, version, kind, gvr)
	return
}

func isClusterWideResource(resource string) bool {
	if strings.HasPrefix(resource, "cluster") || resource == "customresourcedefinitions" || resource == "namespaces" {
		return true
	}
	return false
}

func createObject(client dynamic.Interface, fp string) (err error) {
	fmt.Printf("Reading %s\n", fp)
	// YAML -> Unstructured (through JSON)
	var yConfigMap, jConfigMap []byte
	if yConfigMap, err = os.ReadFile(fp); err != nil {
		err = errors.Wrapf(err, "os.ReadFile failed")
		return
	}
	jConfigMap, err = yaml.ToJSON(yConfigMap)
	if err != nil {
		err = errors.Wrapf(err, "yaml.ToJSON failed")
		return
	}

	var object runtime.Object
	object, err = runtime.Decode(unstructured.UnstructuredJSONScheme, jConfigMap)
	if err != nil {
		err = errors.Wrapf(err, "runtime.Decode failed")
		return
	}

	var uConfigMaps []unstructured.Unstructured
	switch t := object.(type) {
	case *unstructured.Unstructured:
		uConfigMaps = append(uConfigMaps, *t)
	case *unstructured.UnstructuredList:
		uConfigMaps = t.Items
	default:
		err = errors.Wrapf(err, "unstructured.Unstructured or unstructured.UnstructuredList expected")
		return
	}
	for _, uConfigMap := range uConfigMaps {
		var apiVersion, kind, namespace string
		apiVersion = uConfigMap.GetAPIVersion()
		kind = uConfigMap.GetKind()
		gv, _ := schema.ParseGroupVersion(apiVersion)
		var gvr schema.GroupVersionResource
		if gvr, err = kind2Resource(gv.Group, gv.Version, kind); err != nil {
			return
		}
		if isClusterWideResource(gvr.Resource) {
			_, err = client.Resource(gvr).Create(context.TODO(), &uConfigMap, metav1.CreateOptions{})
		} else {
			namespace = uConfigMap.GetNamespace()
			if namespace == "" {
				namespace = apiv1.NamespaceDefault
			}
			_, err = client.Resource(gvr).Namespace(namespace).Create(context.TODO(), &uConfigMap, metav1.CreateOptions{})
		}
		if err != nil {
			if apiErrors.IsAlreadyExists(err) {
				err = nil
				fmt.Printf("Object already exist: namespace %s, name %s, resource %s\n", namespace, uConfigMap.GetName(), gvr.Resource)
				return
			} else {
				err = errors.Wrapf(err, "client.Resource.Namespace.Create %+v failed", gvr)
				return
			}
		}
		fmt.Printf("Created object: namespace %s, name %s, resource %s\n", namespace, uConfigMap.GetName(), gvr.Resource)
	}
	return
}

func main() {
	// Initialize client
	var kubeconfig *string
	if home := homedir.HomeDir(); home != "" {
		kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
	} else {
		kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
	}
	pth := flag.String("f", ".", "path of manifest file or directory")
	flag.Parse()

	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		panic(err)
	}
	client, err := dynamic.NewForConfig(config)
	if err != nil {
		panic(err)
	}

	// Get manifest list
	var fi fs.FileInfo
	if fi, err = os.Stat(*pth); err != nil {
		panic(err)
	}
	var fps []string
	if fi.IsDir() {
		var des []fs.DirEntry
		if des, err = os.ReadDir(*pth); err != nil {
			panic(err)
		}
		for _, de := range des {
			if de.IsDir() {
				continue
			}
			fn := de.Name()
			ext := filepath.Ext(fn)
			if ext != ".yaml" && ext != "yml" {
				continue
			}
			fps = append(fps, filepath.Join(*pth, fn))
		}
		sort.Strings(fps)
	} else {
		fps = append(fps, *pth)
	}

	// Create resource for each manifest
	for _, fp := range fps {
		if err = createObject(client, fp); err != nil {
			fmt.Println(err)
		}
	}
}

[iximiuz]

It's good that you figured it out. At the same time, it seems like you're reinventing the wheel. The module k8s.io/cli-runtime does pretty much the same for you out of the box.

Also, beware that isClusterWideResource() is not the most reliable way of telling if a resource is namespaced or cluster-wide. Every APIResource object has a dedicated Namespaced bool flag for that.