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.