Plugin System (kubernetes-retired#791)

This is an initial implementation of the plugin system kubernetes-retired#509 as proposed in kubernetes-retired#751. Not all but most of knobs mentioned in the proposal except pre/post-cluster-creation validations are implemented.

Basically, it allows the user to define a set of customizations to various aspects of resources created and managed by kube-aws as a "kube-aws plugin" and reuse it.
The set of customizations is defined in a separate file other than a `cluster.yaml` for reusability.

More concretely, provide `<your project root>/plugins/<your-plugin-name>/plugin.yaml` like seen in test/integration/plugin_test.go to extend a kube-aws cluster from many aspects including:

- additional iam policy statements per node role(worker/controller/etcd)
- additional cfn stack template resources per stack(root/control-plane/node-pool)
- additional systemd units/custom files per node role(worker/controller/etcd)
- additional kubelet feature gates for worker kubelets
- additional node labels for worker/controller kubelets

and so on.

The new plugin system is not used to implement core features of kube-aws yet. Therefore, I believe we don't need to worry much about breaking things via this change.

At least one core feature implemented as a plugin is planned in the next version of kube-aws v0.9.9, as noted in our roadmap.

Changes:

* Plugin System: Add support for node labels

* Plugin System: Add support for feature gates

* plugin-system: Add support for k8s manifests and helm releases

* plugin-system: Add support for kube-apiserver server options

* plugin-system: Add support for custom files

* plugin-system: Add support for custom IAM policy statements

* Rename plugin/api to plugin/pluginapi to better differentiate what the api is for

* Move the test helper for plugin to a seperate go file

* Extract a type representing the file uploaded to a kube-aws node into a separate go file

* plugin-system: Seperate logics from api

* plugin-system: Separate cluster extensions by plugins from cluster and plugins

* plugin-system: More separation of api and logic

* plugin-system: Move apply-kube-aws-plugins script for easier merging with master

* plugin-system: Rename pluginapi to pluginmodel

* plugin-system: Remove unused types and files

* plugin-system: Comment about `values` in plugin.yaml

* Reliability improvement to cloud-config-controller
* Fix occasional kube-node-label, cfn-signal errors
* Fix install-kube-system and apply-kube-aws-plugins services to better scheduled in order without spamming journal

core/controlplane/cluster/cluster.go

@@ -13,6 +13,8 @@ import (
 	"github.com/kubernetes-incubator/kube-aws/cfnstack"
 	"github.com/kubernetes-incubator/kube-aws/core/controlplane/config"
 	"github.com/kubernetes-incubator/kube-aws/model"
+	"github.com/kubernetes-incubator/kube-aws/plugin/clusterextension"
+	"github.com/kubernetes-incubator/kube-aws/plugin/pluginmodel"
 )
 
 // VERSION set by build script
@@ -119,21 +121,56 @@ func (c *ClusterRef) validateExistingVPCState(ec2Svc ec2Service) error {
 	return nil
 }
 
-func NewCluster(cfg *config.Cluster, opts config.StackTemplateOptions, awsDebug bool) (*Cluster, error) {
-	cluster := NewClusterRef(cfg, awsDebug)
+func NewCluster(cfg *config.Cluster, opts config.StackTemplateOptions, plugins []*pluginmodel.Plugin, awsDebug bool) (*Cluster, error) {
+	clusterRef := NewClusterRef(cfg, awsDebug)
 	// TODO Do this in a cleaner way e.g. in config.go
-	cluster.KubeResourcesAutosave.S3Path = model.NewS3Folders(opts.S3URI, cluster.ClusterName).ClusterBackups().Path()
-	stackConfig, err := cluster.StackConfig(opts)
+	clusterRef.KubeResourcesAutosave.S3Path = model.NewS3Folders(opts.S3URI, clusterRef.ClusterName).ClusterBackups().Path()
+
+	stackConfig, err := clusterRef.StackConfig(opts, plugins)
 	if err != nil {
 		return nil, err
 	}
 
 	c := &Cluster{
-		ClusterRef:  cluster,
+		ClusterRef:  clusterRef,
 		StackConfig: stackConfig,
 	}
 
+	// Notes:
+	// * `c.StackConfig.CustomSystemdUnits` results in an `ambiguous selector ` error
+	// * `c.Controller.CustomSystemdUnits = controllerUnits` and `c.ClusterRef.Controller.CustomSystemdUnits = controllerUnits` results in modifying invisible/duplicate CustomSystemdSettings
+	extras := clusterextension.NewExtrasFromPlugins(plugins, c.PluginConfigs)
+
+	extraStack, err := extras.ControlPlaneStack()
+	if err != nil {
+		return nil, fmt.Errorf("failed to load control-plane stack extras from plugins: %v", err)
+	}
+	c.StackConfig.ExtraCfnResources = extraStack.Resources
+
+	extraController, err := extras.Controller()
+	if err != nil {
+		return nil, fmt.Errorf("failed to load controller node extras from plugins: %v", err)
+	}
+	c.StackConfig.Config.APIServerFlags = append(c.StackConfig.Config.APIServerFlags, extraController.APIServerFlags...)
+	c.StackConfig.Config.APIServerVolumes = append(c.StackConfig.Config.APIServerVolumes, extraController.APIServerVolumes...)
+	c.StackConfig.Controller.CustomSystemdUnits = append(c.StackConfig.Controller.CustomSystemdUnits, extraController.SystemdUnits...)
+	c.StackConfig.Controller.CustomFiles = append(c.StackConfig.Controller.CustomFiles, extraController.Files...)
+	c.StackConfig.Controller.IAMConfig.Policy.Statements = append(c.StackConfig.Controller.IAMConfig.Policy.Statements, extraController.IAMPolicyStatements...)
+
+	for k, v := range extraController.NodeLabels {
+		c.StackConfig.Controller.NodeLabels[k] = v
+	}
+
+	extraEtcd, err := extras.Etcd()
+	if err != nil {
+		return nil, fmt.Errorf("failed to load controller node extras from plugins: %v", err)
+	}
+	c.StackConfig.Etcd.CustomSystemdUnits = append(c.StackConfig.Etcd.CustomSystemdUnits, extraEtcd.SystemdUnits...)
+	c.StackConfig.Etcd.CustomFiles = append(c.StackConfig.Etcd.CustomFiles, extraEtcd.Files...)
+	c.StackConfig.Etcd.IAMConfig.Policy.Statements = append(c.StackConfig.Etcd.IAMConfig.Policy.Statements, extraEtcd.IAMPolicyStatements...)
+
 	c.assets, err = c.buildAssets()
+
 	return c, err
 }
 

core/controlplane/cluster/cluster_test.go

@@ -18,6 +18,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/kubernetes-incubator/kube-aws/plugin/pluginmodel"
 	yaml "gopkg.in/yaml.v2"
 )
 
@@ -477,7 +478,7 @@ stackTags:
 				S3URI: "s3://test-bucket/foo/bar",
 			}
 
-			cluster, err := NewCluster(clusterConfig, stackTemplateOptions, false)
+			cluster, err := NewCluster(clusterConfig, stackTemplateOptions, []*pluginmodel.Plugin{}, false)
 			if !assert.NoError(t, err) {
 				return
 			}
@@ -741,7 +742,7 @@ func newDefaultClusterWithDeps(opts config.StackTemplateOptions) (*Cluster, erro
 	if err := cluster.Load(); err != nil {
 		return &Cluster{}, err
 	}
-	return NewCluster(cluster, opts, false)
+	return NewCluster(cluster, opts, []*pluginmodel.Plugin{}, false)
 }
 
 func TestRenderStackTemplate(t *testing.T) {

core/controlplane/config/config.go

@@ -6,10 +6,12 @@ package config
 //go:generate gofmt -w files.go
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net"
+	"path/filepath"
 	"regexp"
 	"sort"
 	"strings"
@@ -21,6 +23,8 @@ import (
 	"github.com/kubernetes-incubator/kube-aws/model"
 	"github.com/kubernetes-incubator/kube-aws/model/derived"
 	"github.com/kubernetes-incubator/kube-aws/netutil"
+	"github.com/kubernetes-incubator/kube-aws/node"
+	"github.com/kubernetes-incubator/kube-aws/plugin/pluginmodel"
 	yaml "gopkg.in/yaml.v2"
 )
 
@@ -141,6 +145,7 @@ func NewDefaultCluster() *Cluster {
 			KubeReschedulerImage:               model.Image{Repo: "gcr.io/google-containers/rescheduler", Tag: "v0.3.1", RktPullDocker: false},
 			DnsMasqMetricsImage:                model.Image{Repo: "gcr.io/google_containers/k8s-dns-sidecar-amd64", Tag: "1.14.4", RktPullDocker: false},
 			ExecHealthzImage:                   model.Image{Repo: "gcr.io/google_containers/exechealthz-amd64", Tag: "1.2", RktPullDocker: false},
+			HelmImage:                          model.Image{Repo: "quay.io/kube-aws/helm", Tag: "v2.5.1", RktPullDocker: false},
 			HeapsterImage:                      model.Image{Repo: "gcr.io/google_containers/heapster", Tag: "v1.4.0", RktPullDocker: false},
 			AddonResizerImage:                  model.Image{Repo: "gcr.io/google_containers/addon-resizer", Tag: "2.0", RktPullDocker: false},
 			KubeDashboardImage:                 model.Image{Repo: "gcr.io/google_containers/kubernetes-dashboard-amd64", Tag: "v1.6.3", RktPullDocker: false},
@@ -221,7 +226,7 @@ func ConfigFromBytes(data []byte) (*Config, error) {
 	if err != nil {
 		return nil, err
 	}
-	cfg, err := c.Config()
+	cfg, err := c.Config([]*pluginmodel.Plugin{})
 	if err != nil {
 		return nil, err
 	}
@@ -450,6 +455,7 @@ type DeploymentSettings struct {
 	KubeReschedulerImage               model.Image `yaml:"kubeReschedulerImage,omitempty"`
 	DnsMasqMetricsImage                model.Image `yaml:"dnsMasqMetricsImage,omitempty"`
 	ExecHealthzImage                   model.Image `yaml:"execHealthzImage,omitempty"`
+	HelmImage                          model.Image `yaml:"helmImage,omitempty"`
 	HeapsterImage                      model.Image `yaml:"heapsterImage,omitempty"`
 	AddonResizerImage                  model.Image `yaml:"addonResizerImage,omitempty"`
 	KubeDashboardImage                 model.Image `yaml:"kubeDashboardImage,omitempty"`
@@ -487,20 +493,22 @@ type FlannelSettings struct {
 	PodCIDR string `yaml:"podCIDR,omitempty"`
 }
 
+// Cluster is the container of all the configurable parameters of a kube-aws cluster, customizable via cluster.yaml
 type Cluster struct {
 	KubeClusterSettings    `yaml:",inline"`
 	DeploymentSettings     `yaml:",inline"`
 	DefaultWorkerSettings  `yaml:",inline"`
 	ControllerSettings     `yaml:",inline"`
 	EtcdSettings           `yaml:",inline"`
 	FlannelSettings        `yaml:",inline"`
-	AdminAPIEndpointName   string `yaml:"adminAPIEndpointName,omitempty"`
-	ServiceCIDR            string `yaml:"serviceCIDR,omitempty"`
-	CreateRecordSet        bool   `yaml:"createRecordSet,omitempty"`
-	RecordSetTTL           int    `yaml:"recordSetTTL,omitempty"`
-	TLSCADurationDays      int    `yaml:"tlsCADurationDays,omitempty"`
-	TLSCertDurationDays    int    `yaml:"tlsCertDurationDays,omitempty"`
-	HostedZoneID           string `yaml:"hostedZoneId,omitempty"`
+	AdminAPIEndpointName   string              `yaml:"adminAPIEndpointName,omitempty"`
+	ServiceCIDR            string              `yaml:"serviceCIDR,omitempty"`
+	CreateRecordSet        bool                `yaml:"createRecordSet,omitempty"`
+	RecordSetTTL           int                 `yaml:"recordSetTTL,omitempty"`
+	TLSCADurationDays      int                 `yaml:"tlsCADurationDays,omitempty"`
+	TLSCertDurationDays    int                 `yaml:"tlsCertDurationDays,omitempty"`
+	HostedZoneID           string              `yaml:"hostedZoneId,omitempty"`
+	PluginConfigs          model.PluginConfigs `yaml:"kubeAwsPlugins,omitempty"`
 	ProvidedEncryptService EncryptService
 	// SSHAccessAllowedSourceCIDRs is network ranges of sources you'd like SSH accesses to be allowed from, in CIDR notation
 	SSHAccessAllowedSourceCIDRs model.CIDRRanges       `yaml:"sshAccessAllowedSourceCIDRs,omitempty"`
@@ -711,8 +719,22 @@ func (c KubeClusterSettings) K8sNetworkPlugin() string {
 	return "cni"
 }
 
-func (c Cluster) Config() (*Config, error) {
-	config := Config{Cluster: c}
+func (c Cluster) Config(extra ...[]*pluginmodel.Plugin) (*Config, error) {
+	pluginMap := map[string]*pluginmodel.Plugin{}
+	plugins := []*pluginmodel.Plugin{}
+	if len(extra) > 0 {
+		plugins = extra[0]
+		for _, p := range plugins {
+			pluginMap[p.SettingKey()] = p
+		}
+	}
+
+	config := Config{
+		Cluster:          c,
+		KubeAwsPlugins:   pluginMap,
+		APIServerFlags:   pluginmodel.APIServerFlags{},
+		APIServerVolumes: pluginmodel.APIServerVolumes{},
+	}
 
 	if c.AmiId == "" {
 		var err error
@@ -784,11 +806,18 @@ type StackTemplateOptions struct {
 	SkipWait              bool
 }
 
-func (c Cluster) StackConfig(opts StackTemplateOptions) (*StackConfig, error) {
+func (c Cluster) StackConfig(opts StackTemplateOptions, extra ...[]*pluginmodel.Plugin) (*StackConfig, error) {
+	plugins := []*pluginmodel.Plugin{}
+	if len(extra) > 0 {
+		plugins = extra[0]
+	}
+
 	var err error
-	stackConfig := StackConfig{}
+	stackConfig := StackConfig{
+		ExtraCfnResources: map[string]interface{}{},
+	}
 
-	if stackConfig.Config, err = c.Config(); err != nil {
+	if stackConfig.Config, err = c.Config(plugins); err != nil {
 		return nil, err
 	}
 
@@ -829,15 +858,23 @@ func (c Cluster) StackConfig(opts StackTemplateOptions) (*StackConfig, error) {
 	return &stackConfig, nil
 }
 
+// Config contains configuration parameters available when rendering userdata injected into a controller or an etcd node from golang text templates
 type Config struct {
 	Cluster
 
 	AdminAPIEndpoint derived.APIEndpoint
 	APIEndpoints     derived.APIEndpoints
 
+	// EtcdNodes is the golang-representation of etcd nodes, which is used to differentiate unique etcd nodes
+	// This is used to simplify templating of the control-plane stack template.
 	EtcdNodes []derived.EtcdNode
 
 	AssetsConfig *CompactAssets
+
+	KubeAwsPlugins map[string]*pluginmodel.Plugin
+
+	APIServerVolumes pluginmodel.APIServerVolumes
+	APIServerFlags   pluginmodel.APIServerFlags
 }
 
 // StackName returns the logical name of a CloudFormation stack resource in a root stack template
@@ -1460,6 +1497,128 @@ func (c *Config) ManagedELBLogicalNames() []string {
 	return c.APIEndpoints.ManagedELBLogicalNames()
 }
 
+type kubernetesManifestPlugin struct {
+	Manifests []pluggedInKubernetesManifest
+}
+
+func (p kubernetesManifestPlugin) ManifestListFile() node.UploadedFile {
+	paths := []string{}
+	for _, m := range p.Manifests {
+		paths = append(paths, m.ManifestFile.Path)
+	}
+	bytes := []byte(strings.Join(paths, "\n"))
+	return node.UploadedFile{
+		Path:    p.listFilePath(),
+		Content: node.NewUploadedFileContent(bytes),
+	}
+}
+
+func (p kubernetesManifestPlugin) listFilePath() string {
+	return "/srv/kube-aws/plugins/kubernetes-manifests"
+}
+
+func (p kubernetesManifestPlugin) Directory() string {
+	return filepath.Dir(p.listFilePath())
+}
+
+type pluggedInKubernetesManifest struct {
+	ManifestFile node.UploadedFile
+}
+
+type helmReleasePlugin struct {
+	Releases []pluggedInHelmRelease
+}
+
+func (p helmReleasePlugin) ReleaseListFile() node.UploadedFile {
+	paths := []string{}
+	for _, r := range p.Releases {
+		paths = append(paths, r.ReleaseFile.Path)
+	}
+	bytes := []byte(strings.Join(paths, "\n"))
+	return node.UploadedFile{
+		Path:    p.listFilePath(),
+		Content: node.NewUploadedFileContent(bytes),
+	}
+}
+
+func (p helmReleasePlugin) listFilePath() string {
+	return "/srv/kube-aws/plugins/helm-releases"
+}
+
+func (p helmReleasePlugin) Directory() string {
+	return filepath.Dir(p.listFilePath())
+}
+
+type pluggedInHelmRelease struct {
+	ValuesFile  node.UploadedFile
+	ReleaseFile node.UploadedFile
+}
+
+func (c *Config) KubernetesManifestPlugin() kubernetesManifestPlugin {
+	manifests := []pluggedInKubernetesManifest{}
+	for pluginName, _ := range c.PluginConfigs {
+		plugin, ok := c.KubeAwsPlugins[pluginName]
+		if !ok {
+			panic(fmt.Errorf("Plugin %s is requested but not loaded. Probably a typo in the plugin name inside cluster.yaml?", pluginName))
+		}
+		for _, manifestConfig := range plugin.Configuration.Kubernetes.Manifests {
+			bytes := []byte(manifestConfig.Contents.Inline)
+			m := pluggedInKubernetesManifest{
+				ManifestFile: node.UploadedFile{
+					Path:    filepath.Join("/srv/kube-aws/plugins", plugin.Metadata.Name, manifestConfig.Name),
+					Content: node.NewUploadedFileContent(bytes),
+				},
+			}
+			manifests = append(manifests, m)
+		}
+	}
+	p := kubernetesManifestPlugin{
+		Manifests: manifests,
+	}
+	return p
+}
+
+func (c *Config) HelmReleasePlugin() helmReleasePlugin {
+	releases := []pluggedInHelmRelease{}
+	for pluginName, _ := range c.PluginConfigs {
+		plugin := c.KubeAwsPlugins[pluginName]
+		for _, releaseConfig := range plugin.Configuration.Helm.Releases {
+			valuesFilePath := filepath.Join("/srv/kube-aws/plugins", plugin.Metadata.Name, "helm", "releases", releaseConfig.Name, "values.yaml")
+			valuesFileContent, err := json.Marshal(releaseConfig.Values)
+			if err != nil {
+				panic(fmt.Errorf("Unexpected error in HelmReleasePlugin: %v", err))
+			}
+			releaseFileData := map[string]interface{}{
+				"values": map[string]string{
+					"file": valuesFilePath,
+				},
+				"chart": map[string]string{
+					"name":    releaseConfig.Name,
+					"version": releaseConfig.Version,
+				},
+			}
+			releaseFilePath := filepath.Join("/srv/kube-aws/plugins", plugin.Metadata.Name, "helm", "releases", releaseConfig.Name, "release.json")
+			releaseFileContent, err := json.Marshal(releaseFileData)
+			if err != nil {
+				panic(fmt.Errorf("Unexpected error in HelmReleasePlugin: %v", err))
+			}
+			r := pluggedInHelmRelease{
+				ValuesFile: node.UploadedFile{
+					Path:    valuesFilePath,
+					Content: node.NewUploadedFileContent(valuesFileContent),
+				},
+				ReleaseFile: node.UploadedFile{
+					Path:    releaseFilePath,
+					Content: node.NewUploadedFileContent(releaseFileContent),
+				},
+			}
+			releases = append(releases, r)
+		}
+	}
+	p := helmReleasePlugin{}
+	return p
+}
+
 func WithTrailingDot(s string) string {
 	if s == "" {
 		return s

core/controlplane/config/stack_config.go

@@ -9,12 +9,14 @@ import (
 	"github.com/kubernetes-incubator/kube-aws/model"
 )
 
+// StackConfig contains configuration parameters available when rendering CFN stack template from golang text templates
 type StackConfig struct {
 	*Config
 	StackTemplateOptions
 	UserDataController    model.UserData
 	UserDataEtcd          model.UserData
 	ControllerSubnetIndex int
+	ExtraCfnResources     map[string]interface{}
 }
 
 func (c *StackConfig) s3Folders() model.S3Folders {