Merge pull request #207 from ibuildthecloud/master

Move packages from rancher/rancher to norman

build.go

@@ -0,0 +1,243 @@
+package norman
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/rancher/norman/api"
+	"github.com/rancher/norman/controller"
+	"github.com/rancher/norman/leader"
+	"github.com/rancher/norman/pkg/kwrapper/k3s"
+	"github.com/rancher/norman/pkg/kwrapper/k8s"
+	"github.com/rancher/norman/pkg/remotedialer"
+	"github.com/rancher/norman/store/crd"
+	"github.com/rancher/norman/store/proxy"
+	"github.com/rancher/norman/types"
+	"github.com/sirupsen/logrus"
+	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	"k8s.io/client-go/kubernetes"
+)
+
+type serverContextKey struct{}
+
+func GetServer(ctx context.Context) *Server {
+	return ctx.Value(serverContextKey{}).(*Server)
+}
+
+func (c *Config) Build(ctx context.Context, opts Options) (context.Context, *Server, error) {
+	var (
+		err      error
+		starters []controller.Starter
+	)
+
+	if c.Name == "" {
+		return ctx, nil, errors.New("Name must be set on norman.Config")
+	}
+
+	r := &Runtime{
+		AllSchemas: types.NewSchemas(),
+	}
+
+	server := &Server{
+		Config:  c,
+		Runtime: r,
+	}
+
+	ctx = context.WithValue(ctx, serverContextKey{}, server)
+
+	ctx, err = c.defaults(ctx, r, opts)
+	if err != nil {
+		return ctx, nil, err
+	}
+
+	for _, schema := range c.Schemas {
+		r.AllSchemas.AddSchemas(schema)
+	}
+
+	c.createCRDs(ctx, r)
+
+	ctx, starters, err = c.clients(ctx, r)
+	if err != nil {
+		return ctx, nil, err
+	}
+
+	if c.GlobalSetup != nil {
+		ctx, err = c.GlobalSetup(ctx)
+		if err != nil {
+			return ctx, nil, err
+		}
+	}
+
+	if err := c.registerControllers(ctx, c.PerServerControllers); err != nil {
+		return ctx, nil, err
+	}
+
+	if !opts.DisableControllers {
+		go c.masterControllers(ctx, r, opts)
+	}
+
+	if !c.DisableAPI {
+		if err := c.apiServer(ctx, r); err != nil {
+			return ctx, nil, err
+		}
+	}
+
+	if c.PreStart != nil {
+		if err := c.PreStart(ctx); err != nil {
+			return ctx, nil, err
+		}
+	}
+
+	return ctx, server, controller.SyncThenStart(ctx, c.Threadiness, starters...)
+}
+
+func (c *Config) apiServer(ctx context.Context, r *Runtime) error {
+	if c.K3s.RemoteDialerAuthorizer != nil {
+		r.K3sTunnelServer = remotedialer.New(c.K3s.RemoteDialerAuthorizer, remotedialer.DefaultErrorWriter)
+	}
+
+	server := api.NewAPIServer()
+	if err := server.AddSchemas(r.AllSchemas); err != nil {
+		return err
+	}
+
+	r.APIHandler = server
+
+	if c.APISetup != nil {
+		if err := c.APISetup(ctx, server); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Config) registerControllers(ctx context.Context, controllers []ControllerRegister) error {
+	for _, controller := range controllers {
+		if err := controller(ctx); err != nil {
+			return fmt.Errorf("failed to start controller: %v", err)
+		}
+	}
+
+	return nil
+}
+
+func (c *Config) masterControllers(ctx context.Context, r *Runtime, opts Options) {
+	leader.RunOrDie(ctx, c.Name, c.K8sClient, func(ctx context.Context) {
+		var (
+			err      error
+			starters []controller.Starter
+		)
+
+		ctx, starters, err = c.clients(ctx, r)
+		if err != nil {
+			logrus.Fatalf("failed to create master clients: %v", err)
+		}
+
+		if c.MasterSetup != nil {
+			ctx, err = c.MasterSetup(ctx)
+			if err != nil {
+				logrus.Fatalf("failed to setup master: %v", err)
+			}
+		}
+
+		err = c.registerControllers(ctx, c.MasterControllers)
+		if err != nil {
+			logrus.Fatalf("failed to register master controllers: %v", err)
+		}
+
+		if err := controller.SyncThenStart(ctx, c.Threadiness, starters...); err != nil {
+			logrus.Fatalf("failed to start master controllers: %v", err)
+		}
+
+		<-ctx.Done()
+	})
+}
+
+func (c *Config) defaults(ctx context.Context, r *Runtime, opts Options) (context.Context, error) {
+	var (
+		err error
+	)
+
+	if c.Threadiness <= 0 {
+		c.Threadiness = 5
+	}
+
+	if c.Config == nil {
+		mode := "auto"
+		if opts.K8sMode != "" {
+			mode = opts.K8sMode
+		}
+
+		envConfig := os.Getenv("KUBECONFIG")
+		if c.IgnoredKubConfigEnv {
+			envConfig = ""
+		}
+
+		if c.K3s.DataDir != "" && c.K3s.RemoteDialerAuthorizer != nil {
+			ctx, r.K3sServerConfig, r.K3sTunnelServer, err = k3s.NewConfig(ctx, c.K3s.DataDir, c.K3s.RemoteDialerAuthorizer)
+			if err != nil {
+				return ctx, err
+			}
+		}
+
+		r.Embedded, ctx, c.Config, err = k8s.GetConfig(ctx, mode, envConfig)
+		if err != nil {
+			return ctx, err
+		}
+	}
+
+	if c.ClientGetter == nil {
+		cg, err := proxy.NewClientGetterFromConfig(*c.Config)
+		if err != nil {
+			return ctx, err
+		}
+		c.ClientGetter = cg
+	}
+
+	return ctx, nil
+}
+
+func (c *Config) createCRDs(ctx context.Context, r *Runtime) {
+	factory := &crd.Factory{ClientGetter: c.ClientGetter}
+
+	for version, types := range c.CRDs {
+		factory.BatchCreateCRDs(ctx, c.CRDStorageContext, r.AllSchemas, version, types...)
+	}
+
+	factory.BatchWait()
+}
+
+func (c *Config) clients(ctx context.Context, r *Runtime) (context.Context, []controller.Starter, error) {
+	var (
+		starter  controller.Starter
+		starters []controller.Starter
+		err      error
+	)
+
+	if c.K8sClient == nil {
+		c.K8sClient, err = kubernetes.NewForConfig(c.Config)
+		if err != nil {
+			return ctx, nil, err
+		}
+	}
+
+	if c.APIExtClient == nil {
+		c.APIExtClient, err = clientset.NewForConfig(c.Config)
+		if err != nil {
+			return ctx, nil, err
+		}
+	}
+
+	for _, clientFactory := range c.Clients {
+		ctx, starter, err = clientFactory(ctx, *c.Config)
+		if err != nil {
+			return ctx, nil, err
+		}
+		starters = append(starters, starter)
+	}
+
+	return ctx, starters, nil
+}

controller/generic_controller.go

@@ -44,6 +44,7 @@ func init() {
 type HandlerFunc func(key string) error
 
 type GenericController interface {
+	SetThreadinessOverride(count int)
 	Informer() cache.SharedIndexInformer
 	AddHandler(name string, handler HandlerFunc)
 	HandlerCount() int
@@ -65,12 +66,13 @@ type handlerDef struct {
 
 type genericController struct {
 	sync.Mutex
-	informer cache.SharedIndexInformer
-	handlers []handlerDef
-	queue    workqueue.RateLimitingInterface
-	name     string
-	running  bool
-	synced   bool
+	threadinessOverride int
+	informer            cache.SharedIndexInformer
+	handlers            []handlerDef
+	queue               workqueue.RateLimitingInterface
+	name                string
+	running             bool
+	synced              bool
 }
 
 func NewGenericController(name string, genericClient Backend) GenericController {
@@ -94,6 +96,10 @@ func NewGenericController(name string, genericClient Backend) GenericController
 	}
 }
 
+func (g *genericController) SetThreadinessOverride(count int) {
+	g.threadinessOverride = count
+}
+
 func (g *genericController) HandlerCount() int {
 	return len(g.handlers)
 }
@@ -163,6 +169,9 @@ func (g *genericController) Start(ctx context.Context, threadiness int) error {
 	}
 
 	if !g.running {
+		if g.threadinessOverride > 0 {
+			threadiness = g.threadinessOverride
+		}
 		go g.run(ctx, threadiness)
 	}
 

generator/generator.go

@@ -2,7 +2,6 @@ package generator
 
 import (
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net/http"
 	"os"
@@ -504,14 +503,6 @@ func deepCopyGen(workDir, pkg string) error {
 		})
 }
 
-type noInitGenerator struct {
-	generator.Generator
-}
-
-func (n *noInitGenerator) Init(*generator.Context, io.Writer) error {
-	return nil
-}
-
 func isObjectOrList(t *gengotypes.Type) bool {
 	for _, member := range t.Members {
 		if member.Embedded && (member.Name == "ObjectMeta" || member.Name == "ListMeta") {

generator/k8s_client_template.go

@@ -13,6 +13,8 @@ import (
 	"k8s.io/client-go/rest"
 )
 
+type contextKeyType struct{}
+
 type Interface interface {
 	RESTClient() rest.Interface
 	controller.Starter
@@ -28,6 +30,20 @@ type Client struct {
 	{{.ID}}Controllers map[string]{{.CodeName}}Controller{{end}}
 }
 
+func Factory(ctx context.Context, config rest.Config) (context.Context, controller.Starter, error) {
+	c, err := NewForConfig(config)
+	if err != nil {
+		return ctx, nil, err
+	}
+
+	return context.WithValue(ctx, contextKeyType{}, c), c, nil
+}
+
+func From(ctx context.Context) Interface {
+	return ctx.Value(contextKeyType{}).(Interface)
+}
+
+
 func NewForConfig(config rest.Config) (Interface, error) {
 	if config.NegotiatedSerializer == nil {
 		config.NegotiatedSerializer = dynamic.NegotiatedSerializer