Move YAML decode logic into provider and improve default Helm namespaces

CHANGELOG.md

@@ -3,6 +3,9 @@
 ### Improvements
 
 -   Improve namespaced Kind check. (https://github.com/pulumi/pulumi-kubernetes/pull/947).
+-   Add helm template `apiVersions` flag. (https://github.com/pulumi/pulumi-kubernetes/pull/894)
+-   Move YAML decode logic into provider and improve handling of default namespaces for Helm charts. (https://github.com
+/pulumi/pulumi-kubernetes/pull/952).
 
 ### Bug fixes
 
@@ -28,10 +31,6 @@
 -   Fix deprecation warnings and docs. (https://github.com/pulumi/pulumi-kubernetes/pull/929).
 -   Fix projection of array-valued output properties in .NET. (https://github.com/pulumi/pulumi-kubernetes/pull/931)
 
-### Improvements
-
--   Add helm template `apiVersions` flag. (https://github.com/pulumi/pulumi-kubernetes/pull/894)
-
 ## 1.4.1 (December 17, 2019)
 
 ### Bug fixes

pkg/gen/nodejs-templates/helm/v2/helm.ts

@@ -217,7 +217,8 @@ export class Chart extends yaml.CollectionComponentResource {
                         maxBuffer: 512 * 1024 * 1024 // 512 MB
                     },
                 ).toString();
-                return this.parseTemplate(yamlStream, cfg.transformations, cfg.resourcePrefix, configDeps);
+                return this.parseTemplate(
+                    yamlStream, cfg.transformations, cfg.resourcePrefix, configDeps, cfg.namespace);
             } catch (e) {
                 // Shed stack trace, only emit the error.
                 throw new pulumi.RunError(e.toString());
@@ -230,93 +231,25 @@ export class Chart extends yaml.CollectionComponentResource {
     }
 
     parseTemplate(
-        yamlStream: string,
+        text: string,
         transformations: ((o: any, opts: pulumi.CustomResourceOptions) => void)[] | undefined,
         resourcePrefix: string | undefined,
         dependsOn: pulumi.Resource[],
+        defaultNamespace: string | undefined,
     ): pulumi.Output<{ [key: string]: pulumi.CustomResource }> {
-        // NOTE: We must manually split the YAML stream because of js-yaml#456. Perusing the code
-        // and the spec, it looks like a YAML stream is delimited by `^---`, though it is difficult
-        // to know for sure.
-        //
-        // NOTE: We use `{json: true, schema: jsyaml.CORE_SCHEMA}` here so that we conform to Helm's
-        // YAML parsing semantics. Specifically, `json: true` to ensure that a duplicate key
-        // overrides its predecessory, rather than throwing an exception, and `schema:
-        // jsyaml.CORE_SCHEMA` to avoid using additional YAML parsing rules not supported by the
-        // YAML parser used by Kubernetes.
-        const objs = yamlStream.split(/^---/m)
-            .map(yaml => jsyaml.safeLoad(yaml, {json: true, schema: jsyaml.CORE_SCHEMA}))
-            .filter(a => a != null && "kind" in a)
-            .sort(helmSort);
-        return yaml.parse(
+        const promise = pulumi.runtime.invoke(
+            "kubernetes:yaml:decode", {text, defaultNamespace}, {async: true});
+        return pulumi.output(promise).apply<{[key: string]: pulumi.CustomResource}>(p => yaml.parse(
             {
                 resourcePrefix: resourcePrefix,
-                yaml: objs.map(o => jsyaml.safeDump(o)),
+                objs: p.result,
                 transformations: transformations || [],
             },
             { parent: this, dependsOn: dependsOn }
-        );
+        ));
     }
 }
 
-// helmSort is a JavaScript implementation of the Helm Kind sorter[1]. It provides a
-// best-effort topology of Kubernetes kinds, which in most cases should ensure that resources
-// that must be created first, are.
-//
-// [1]: https://github.com/helm/helm/blob/094b97ab5d7e2f6eda6d0ab0f2ede9cf578c003c/pkg/tiller/kind_sorter.go
-/** @ignore */ export function helmSort(a: { kind: string }, b: { kind: string }): number {
-    const installOrder = [
-        "Namespace",
-        "ResourceQuota",
-        "LimitRange",
-        "PodSecurityPolicy",
-        "Secret",
-        "ConfigMap",
-        "StorageClass",
-        "PersistentVolume",
-        "PersistentVolumeClaim",
-        "ServiceAccount",
-        "CustomResourceDefinition",
-        "ClusterRole",
-        "ClusterRoleBinding",
-        "Role",
-        "RoleBinding",
-        "Service",
-        "DaemonSet",
-        "Pod",
-        "ReplicationController",
-        "ReplicaSet",
-        "Deployment",
-        "StatefulSet",
-        "Job",
-        "CronJob",
-        "Ingress",
-        "APIService"
-    ];
-
-    const ordering: { [key: string]: number } = {};
-    installOrder.forEach((_, i) => {
-        ordering[installOrder[i]] = i;
-    });
-
-    const aKind = a["kind"];
-    const bKind = b["kind"];
-
-    if (!(aKind in ordering) && !(bKind in ordering)) {
-        return aKind.localeCompare(bKind);
-    }
-
-    if (!(aKind in ordering)) {
-        return 1;
-    }
-
-    if (!(bKind in ordering)) {
-        return -1;
-    }
-
-    return ordering[aKind] - ordering[bKind];
-}
-
 /**
  * Additional options to customize the fetching of the Helm chart.
  */

pkg/gen/nodejs-templates/yaml.ts.mustache

@@ -87,7 +87,7 @@ import * as outputs from "../types/output";
 
     export interface ConfigOpts {
         /** JavaScript objects representing Kubernetes resources. */
-        objs: any[];
+        objs: pulumi.Input<pulumi.Input<any>[]>;
 
         /**
          * A set of transformations to apply to Kubernetes resource definitions before registering
@@ -120,14 +120,9 @@ import * as outputs from "../types/output";
         resourcePrefix?: string;
     }
 
-    function yamlLoadAll(text: string): any[] {
-        // NOTE[pulumi-kubernetes#501]: Use `loadAll` with `JSON_SCHEMA` here instead of
-        // `safeLoadAll` because the latter is incompatible with `JSON_SCHEMA`. It is
-        // important to use `JSON_SCHEMA` here because the fields of the Kubernetes core
-        // API types are all tagged with `json:`, and they don't deal very well with things
-        // like dates.
-        const jsyaml = require("js-yaml");
-        return jsyaml.loadAll(text, undefined, {schema: jsyaml.JSON_SCHEMA});
+    function yamlLoadAll(text: string): Promise<any[]> {
+        const promise = pulumi.runtime.invoke("kubernetes:yaml:decode", {text}, {async: true});
+        return promise.then(p => p.result);
     }
 
     /** @ignore */ export function parse(
@@ -340,22 +335,13 @@ import * as outputs from "../types/output";
         config: ConfigOpts,
         opts?: pulumi.CustomResourceOptions,
     ):  pulumi.Output<{[key: string]: pulumi.CustomResource}> {
-        const objs: pulumi.Output<{name: string, resource: pulumi.CustomResource}>[] = [];
+        return pulumi.output(config.objs).apply(configObjs => {
+            const objs = configObjs
+                .map(obj => parseYamlObject(obj, config.transformations, config.resourcePrefix, opts))
+                .reduce((array, objs) => (array.concat(...objs)), []);
 
-        for (const obj of config.objs) {
-            const fileObjects: pulumi.Output<{name: string, resource: pulumi.CustomResource}>[] =
-                parseYamlObject(obj, config.transformations, config.resourcePrefix, opts);
-            for (const fileObject of fileObjects) {
-                objs.push(fileObject);
-            }
-        }
-        return pulumi.all(objs).apply(xs => {
-            let resources: {[key: string]: pulumi.CustomResource} = {};
-            for (const x of xs) {
-                resources[x.name] = x.resource
-            }
-
-            return resources;
+            return pulumi.output(objs).apply(objs => objs
+                .reduce((map: {[key: string]: pulumi.CustomResource}, val) => (map[val.name] = val.resource, map), {}))
         });
     }
 

pkg/gen/python-templates/helm/v2/helm.py

@@ -9,7 +9,6 @@
 from typing import Any, Callable, List, Optional, TextIO, Tuple, Union
 
 import pulumi.runtime
-import yaml
 from pulumi_kubernetes.yaml import _parse_yaml_document
 
 
@@ -348,10 +347,13 @@ def _parse_chart(all_config: Tuple[str, Union[ChartOpts, LocalChartOpts], pulumi
     cmd.extend(home_arg)
 
     chart_resources = pulumi.Output.all(cmd, data).apply(_run_helm_cmd)
+    objects = chart_resources.apply(
+        lambda text: pulumi.runtime.invoke('kubernetes:yaml:decode', {
+            'text': text, 'defaultNamespace': config.namespace}).value['result'])
 
     # Parse the manifest and create the specified resources.
-    resources = chart_resources.apply(
-        lambda yaml_str: _parse_yaml_document(yaml.safe_load_all(yaml_str), opts, config.transformations))
+    resources = objects.apply(
+        lambda objects: _parse_yaml_document(objects, opts, config.transformations))
 
     pulumi.Output.all(file, chart_dir, resources).apply(_cleanup_temp_dir)
     return resources
@@ -472,7 +474,7 @@ def get_resource(self, group_version_kind, name, namespace=None) -> pulumi.Outpu
 
         # `id` will either be `${name}` or `${namespace}/${name}`.
         id = pulumi.Output.from_input(name)
-        if namespace != None:
+        if namespace is not None:
             id = pulumi.Output.concat(namespace, '/', name)
 
         resource_id = id.apply(lambda x: f'{group_version_kind}:{x}')

pkg/gen/python-templates/yaml.py.mustache

@@ -8,8 +8,6 @@ from typing import Callable, Dict, List, Optional
 
 import pulumi.runtime
 import requests
-import yaml
-import pulumi_kubernetes
 from pulumi_kubernetes.apiextensions import CustomResource
 
 from . import tables
@@ -25,7 +23,6 @@ class ConfigFile(pulumi.ComponentResource):
     Kubernetes resources contained in this ConfigFile.
     """
 
-
     def __init__(self, name, file_id, opts=None, transformations=None, resource_prefix=None):
         """
         :param str name: A name for a resource.
@@ -63,7 +60,8 @@ class ConfigFile(pulumi.ComponentResource):
         # Note: Unlike NodeJS, Python requires that we "pull" on our futures in order to get them scheduled for
         # execution. In order to do this, we leverage the engine's RegisterResourceOutputs to wait for the
         # resolution of all resources that this YAML document created.
-        self.resources = _parse_yaml_document(yaml.safe_load_all(text), opts, transformations, resource_prefix)
+        __ret__ = pulumi.runtime.invoke('kubernetes:yaml:decode', {'text': text}).value['result']
+        self.resources = _parse_yaml_document(__ret__, opts, transformations, resource_prefix)
         self.register_outputs({"resources": self.resources})
 
     def translate_output_property(self, prop: str) -> str:
@@ -84,12 +82,13 @@ class ConfigFile(pulumi.ComponentResource):
 
         # `id` will either be `${name}` or `${namespace}/${name}`.
         id = pulumi.Output.from_input(name)
-        if namespace != None:
+        if namespace is not None:
             id = pulumi.Output.concat(namespace, '/', name)
 
         resource_id = id.apply(lambda x: f'{group_version_kind}:{x}')
         return resource_id.apply(lambda x: self.resources[x])
 
+
 def _read_url(url: str) -> str:
     response = requests.get(url)
     response.raise_for_status()

pkg/provider/invoke_decode_yaml.go

@@ -0,0 +1,73 @@
+// Copyright 2016-2019, Pulumi Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+	"io"
+	"io/ioutil"
+	"strings"
+
+	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/util/yaml"
+)
+
+// decodeYaml parses a YAML string, and then returns a slice of untyped structs that can be marshalled into
+// Pulumi RPC calls. If a default namespace is specified, set that on the relevant decoded objects.
+func decodeYaml(text, defaultNamespace string, clientSet *clients.DynamicClientSet) ([]interface{}, error) {
+	var resources []unstructured.Unstructured
+
+	dec := yaml.NewYAMLOrJSONDecoder(ioutil.NopCloser(strings.NewReader(text)), 128)
+	for {
+		var value map[string]interface{}
+		if err := dec.Decode(&value); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, err
+		}
+		resource := unstructured.Unstructured{Object: value}
+
+		// Sometimes manifests include empty resources, so skip these.
+		if len(resource.GetKind()) == 0 || len(resource.GetAPIVersion()) == 0 {
+			continue
+		}
+
+		if len(defaultNamespace) > 0 {
+			namespaced, err := clients.IsNamespacedKind(resource.GroupVersionKind(), clientSet)
+			if err != nil {
+				if clients.IsNoNamespaceInfoErr(err) {
+					// Assume resource is namespaced.
+					namespaced = true
+				} else {
+					return nil, err
+				}
+			}
+
+			// Set namespace if resource Kind is namespaced and namespace is not already set.
+			if namespaced && len(resource.GetNamespace()) == 0 {
+				resource.SetNamespace(defaultNamespace)
+			}
+		}
+		resources = append(resources, resource)
+	}
+
+	result := make([]interface{}, len(resources))
+	for _, resource := range resources {
+		result = append(result, resource.Object)
+	}
+
+	return result, nil
+}

pkg/provider/provider.go

@@ -72,6 +72,7 @@ const (
 	streamInvokeList     = "kubernetes:kubernetes:list"
 	streamInvokeWatch    = "kubernetes:kubernetes:watch"
 	streamInvokePodLogs  = "kubernetes:kubernetes:podLogs"
+	invokeDecodeYaml     = "kubernetes:yaml:decode"
 	lastAppliedConfigKey = "kubectl.kubernetes.io/last-applied-configuration"
 	initialApiVersionKey = "__initialApiVersion"
 )
@@ -389,9 +390,45 @@ func (k *kubeProvider) Invoke(ctx context.Context,
 	// inputs may not have resolved yet. Any invoke logic that depends on an active cluster must check
 	// k.clusterUnreachable and handle that condition appropriately.
 
-	// Always fail.
 	tok := req.GetTok()
-	return nil, fmt.Errorf("unknown Invoke type '%s'", tok)
+	label := fmt.Sprintf("%s.Invoke(%s)", k.label(), tok)
+	args, err := plugin.UnmarshalProperties(
+		req.GetArgs(), plugin.MarshalOptions{Label: label, KeepUnknowns: true})
+	if err != nil {
+		return nil, pkgerrors.Wrapf(err, "failed to unmarshal %v args during an Invoke call", tok)
+	}
+
+	switch tok {
+	case invokeDecodeYaml:
+		var text, defaultNamespace string
+		if args["text"].HasValue() {
+			text = args["text"].StringValue()
+		} else {
+			return nil, fmt.Errorf("missing required field 'text' on input: %#v", args)
+		}
+		if args["defaultNamespace"].HasValue() {
+			defaultNamespace = args["defaultNamespace"].StringValue()
+		}
+
+		result, err := decodeYaml(text, defaultNamespace, k.clientSet)
+		if err != nil {
+			return nil, err
+		}
+
+		objProps, err := plugin.MarshalProperties(
+			resource.NewPropertyMapFromMap(map[string]interface{}{"result": result}),
+			plugin.MarshalOptions{
+				Label: label, KeepUnknowns: true, SkipNulls: true,
+			})
+		if err != nil {
+			return nil, err
+		}
+
+		return &pulumirpc.InvokeResponse{Return: objProps}, nil
+
+	default:
+		return nil, fmt.Errorf("unknown Invoke type %q", tok)
+	}
 }
 
 // StreamInvoke dynamically executes a built-in function in the provider. The result is streamed