norman/store/proxy/proxy_store.go

package proxy

import (
	ejson "encoding/json"
	"strings"

	"github.com/rancher/norman/types"
	"github.com/rancher/norman/types/convert"
	"github.com/rancher/norman/types/values"
	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/runtime/serializer/json"
	"k8s.io/apimachinery/pkg/runtime/serializer/streaming"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
	restclientwatch "k8s.io/client-go/rest/watch"
)

type Store struct {
	k8sClient      rest.Interface
	prefix         []string
	group          string
	version        string
	kind           string
	resourcePlural string
}

func NewProxyStore(k8sClient rest.Interface,
	prefix []string, group, version, kind, resourcePlural string) *Store {
	return &Store{
		k8sClient:      k8sClient,
		prefix:         prefix,
		group:          group,
		version:        version,
		kind:           kind,
		resourcePlural: resourcePlural,
	}
}

func (p *Store) ByID(apiContext *types.APIContext, schema *types.Schema, id string) (map[string]interface{}, error) {
	_, result, err := p.byID(apiContext, schema, id)
	return result, err
}

func (p *Store) byID(apiContext *types.APIContext, schema *types.Schema, id string) (string, map[string]interface{}, error) {
	namespace, id := splitID(id)

	req := p.common(namespace, p.k8sClient.Get()).
		Name(id)

	return p.singleResult(schema, req)
}

func (p *Store) List(apiContext *types.APIContext, schema *types.Schema, opt types.QueryOptions) ([]map[string]interface{}, error) {
	namespace := getNamespace(apiContext, opt)

	req := p.common(namespace, p.k8sClient.Get())

	resultList := &unstructured.UnstructuredList{}
	err := req.Do().Into(resultList)
	if err != nil {
		return nil, err
	}

	var result []map[string]interface{}

	for _, obj := range resultList.Items {
		result = append(result, p.fromInternal(schema, obj.Object))
	}

	return result, nil
}

func (p *Store) Watch(apiContext *types.APIContext, schema *types.Schema, opt types.QueryOptions) (chan map[string]interface{}, error) {
	namespace := getNamespace(apiContext, opt)

	req := p.common(namespace, p.k8sClient.Get())
	req.VersionedParams(&metav1.ListOptions{
		Watch: true,
	}, dynamic.VersionedParameterEncoderWithV1Fallback)

	body, err := req.Stream()
	if err != nil {
		return nil, err
	}

	framer := json.Framer.NewFrameReader(body)
	decoder := streaming.NewDecoder(framer, &unstructuredDecoder{})
	watcher := watch.NewStreamWatcher(restclientwatch.NewDecoder(decoder, &unstructuredDecoder{}))

	go func() {
		<-apiContext.Request.Context().Done()
		watcher.Stop()
	}()

	result := make(chan map[string]interface{})
	go func() {
		for event := range watcher.ResultChan() {
			data := event.Object.(*unstructured.Unstructured)
			p.fromInternal(schema, data.Object)
			result <- data.Object
		}
		close(result)
	}()

	return result, nil
}

type unstructuredDecoder struct {
}

func (d *unstructuredDecoder) Decode(data []byte, defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
	if into == nil {
		into = &unstructured.Unstructured{}
	}
	return into, defaults, ejson.Unmarshal(data, &into)
}

func getNamespace(apiContext *types.APIContext, opt types.QueryOptions) string {
	if val, ok := apiContext.SubContext["namespaces"]; ok {
		return convert.ToString(val)
	}

	for _, condition := range opt.Conditions {
		if condition.Field == "namespaceId" && condition.Value != "" {
			return condition.Value
		}
	}

	return ""
}

func (p *Store) Create(apiContext *types.APIContext, schema *types.Schema, data map[string]interface{}) (map[string]interface{}, error) {
	namespace, _ := data["namespaceId"].(string)
	p.toInternal(schema.Mapper, data)

	name, _ := values.GetValueN(data, "metadata", "name").(string)
	if name == "" {
		generated, _ := values.GetValueN(data, "metadata", "generateName").(string)
		if generated == "" {
			values.PutValue(data, strings.ToLower(schema.ID+"-"), "metadata", "generateName")
		}
	}

	req := p.common(namespace, p.k8sClient.Post()).
		Body(&unstructured.Unstructured{
			Object: data,
		})

	_, result, err := p.singleResult(schema, req)
	return result, err
}

func (p *Store) toInternal(mapper types.Mapper, data map[string]interface{}) {
	if mapper != nil {
		mapper.ToInternal(data)
	}

	if p.group == "" {
		data["apiVersion"] = p.version
	} else {
		data["apiVersion"] = p.group + "/" + p.version
	}
	data["kind"] = p.kind
}

func (p *Store) Update(apiContext *types.APIContext, schema *types.Schema, data map[string]interface{}, id string) (map[string]interface{}, error) {
	resourceVersion, existing, err := p.byID(apiContext, schema, id)
	if err != nil {
		return data, nil
	}

	for k, v := range data {
		existing[k] = v
	}

	p.toInternal(schema.Mapper, existing)
	namespace, id := splitID(id)

	values.PutValue(existing, resourceVersion, "metadata", "resourceVersion")

	req := p.common(namespace, p.k8sClient.Put()).
		Body(&unstructured.Unstructured{
			Object: existing,
		}).
		Name(id)

	_, result, err := p.singleResult(schema, req)
	return result, err
}

func (p *Store) Delete(apiContext *types.APIContext, schema *types.Schema, id string) (map[string]interface{}, error) {
	namespace, id := splitID(id)

	prop := metav1.DeletePropagationForeground
	req := p.common(namespace, p.k8sClient.Delete()).
		Body(&metav1.DeleteOptions{
			PropagationPolicy: &prop,
		}).
		Name(id)

	err := req.Do().Error()
	if err != nil {
		return nil, err
	}

	obj, err := p.ByID(apiContext, schema, id)
	if err != nil {
		return nil, nil
	}
	return obj, nil
}

func (p *Store) singleResult(schema *types.Schema, req *rest.Request) (string, map[string]interface{}, error) {
	result := &unstructured.Unstructured{}
	err := req.Do().Into(result)
	if err != nil {
		return "", nil, err
	}

	version := result.GetResourceVersion()
	p.fromInternal(schema, result.Object)
	return version, result.Object, nil
}

func splitID(id string) (string, string) {
	namespace := ""
	parts := strings.SplitN(id, ":", 2)
	if len(parts) == 2 {
		namespace = parts[0]
		id = parts[1]
	}

	return namespace, id
}

func (p *Store) common(namespace string, req *rest.Request) *rest.Request {
	prefix := append([]string{}, p.prefix...)
	if p.group != "" {
		prefix = append(prefix, p.group)
	}
	prefix = append(prefix, p.version)
	req.Prefix(prefix...).
		Resource(p.resourcePlural)

	if namespace != "" {
		req.Namespace(namespace)
	}

	return req
}

func (p *Store) fromInternal(schema *types.Schema, data map[string]interface{}) map[string]interface{} {
	if schema.Mapper != nil {
		schema.Mapper.FromInternal(data)
	}

	return data
}