Merge pull request #221 from ibuildthecloud/master

Random fixes

build.go

@@ -68,6 +68,12 @@ func (c *Config) Build(ctx context.Context, opts *Options) (context.Context, *Se
 		return ctx, nil, err
 	}
 
+	if c.CustomizeSchemas != nil {
+		if err := c.CustomizeSchemas(ctx, c.ClientGetter, r.AllSchemas); err != nil {
+			return ctx, nil, err
+		}
+	}
+
 	if c.GlobalSetup != nil {
 		ctx, err = c.GlobalSetup(ctx)
 		if err != nil {
@@ -80,7 +86,7 @@ func (c *Config) Build(ctx context.Context, opts *Options) (context.Context, *Se
 	}
 
 	if !opts.DisableControllers {
-		go c.masterControllers(ctx, r)
+		go c.masterControllers(ctx, starters, r)
 	}
 
 	if !c.DisableAPI {
@@ -99,7 +105,7 @@ func (c *Config) Build(ctx context.Context, opts *Options) (context.Context, *Se
 }
 
 func (c *Config) apiServer(ctx context.Context, r *Runtime) error {
-	if c.K3s.RemoteDialerAuthorizer != nil {
+	if c.K3s.RemoteDialerAuthorizer != nil && r.K3sTunnelServer == nil {
 		r.K3sTunnelServer = remotedialer.New(c.K3s.RemoteDialerAuthorizer, remotedialer.DefaultErrorWriter)
 	}
 
@@ -129,18 +135,12 @@ func (c *Config) registerControllers(ctx context.Context, controllers []Controll
 	return nil
 }
 
-func (c *Config) masterControllers(ctx context.Context, r *Runtime) {
+func (c *Config) masterControllers(ctx context.Context, starters []controller.Starter, r *Runtime) {
 	leader.RunOrDie(ctx, c.LeaderLockNamespace, 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 {

controller/generic_controller.go

@@ -187,11 +187,9 @@ func (g *genericController) Start(ctx context.Context, threadiness int) error {
 	g.Lock()
 	defer g.Unlock()
 
-	if !g.synced {
 	if err := g.sync(ctx); err != nil {
 		return err
 	}
-	}
 
 	if !g.running {
 		if g.threadinessOverride > 0 {
@@ -202,7 +200,7 @@ func (g *genericController) Start(ctx context.Context, threadiness int) error {
 
 	if g.running {
 		for _, h := range g.handlers {
-			if h.generation != g.generation {
+			if h.generation < g.generation {
 				continue
 			}
 			for _, key := range g.informer.GetStore().ListKeys() {

generator/default.go

@@ -14,7 +14,7 @@ var (
 	baseK8s    = "apis"
 )
 
-func DefaultGenerate(schemas *types.Schemas, pkgPath string, publicAPI bool, foreignTypes map[string]bool) error {
+func DefaultGenerate(schemas *types.Schemas, pkgPath string, publicAPI bool, privateTypes map[string]bool) error {
 	version := getVersion(schemas)
 	group := strings.Split(version.Group, ".")[0]
 
@@ -24,7 +24,7 @@ func DefaultGenerate(schemas *types.Schemas, pkgPath string, publicAPI bool, for
 	}
 	k8sOutputPackage := path.Join(pkgPath, baseK8s, version.Group, version.Version)
 
-	if err := Generate(schemas, foreignTypes, cattleOutputPackage, k8sOutputPackage); err != nil {
+	if err := Generate(schemas, privateTypes, cattleOutputPackage, k8sOutputPackage); err != nil {
 		return err
 	}
 

generator/generator.go

@@ -361,7 +361,7 @@ func GenerateControllerForTypes(version *types.APIVersion, k8sOutputPackage stri
 	return gofmt(baseDir, k8sOutputPackage)
 }
 
-func Generate(schemas *types.Schemas, foreignTypes map[string]bool, cattleOutputPackage, k8sOutputPackage string) error {
+func Generate(schemas *types.Schemas, privateTypes map[string]bool, cattleOutputPackage, k8sOutputPackage string) error {
 	baseDir := args.DefaultSourceTree()
 	cattleDir := path.Join(baseDir, cattleOutputPackage)
 	k8sDir := path.Join(baseDir, k8sOutputPackage)
@@ -382,7 +382,7 @@ func Generate(schemas *types.Schemas, foreignTypes map[string]bool, cattleOutput
 			continue
 		}
 
-		_, backendType := foreignTypes[schema.ID]
+		_, privateType := privateTypes[schema.ID]
 
 		if cattleDir != "" {
 			if err := generateType(cattleDir, schema, schemas); err != nil {
@@ -390,7 +390,7 @@ func Generate(schemas *types.Schemas, foreignTypes map[string]bool, cattleOutput
 			}
 		}
 
-		if backendType ||
+		if privateType ||
 			(contains(schema.CollectionMethods, http.MethodGet) &&
 				!strings.HasPrefix(schema.PkgName, "k8s.io") &&
 				!strings.Contains(schema.PkgName, "/vendor/")) {
@@ -403,7 +403,7 @@ func Generate(schemas *types.Schemas, foreignTypes map[string]bool, cattleOutput
 			}
 		}
 
-		if !backendType {
+		if !privateType {
 			cattleClientTypes = append(cattleClientTypes, schema)
 		}
 	}

pkg/subscribe/handler.go

@@ -109,16 +109,17 @@ func handler(apiContext *types.APIContext) error {
 			if schema != nil {
 				buffer := &bytes.Buffer{}
 				if err := jsonWriter.VersionBody(apiContext, &schema.Version, buffer, item); err != nil {
-					return err
+					cancel()
+					continue
 				}
 
 				if err := writeData(c, header, buffer.Bytes()); err != nil {
-					return err
+					cancel()
 				}
 			}
 		case <-t.C:
 			if err := writeData(c, `{"name":"ping","data":`, []byte("{}")); err != nil {
-				return err
+				cancel()
 			}
 		}
 	}

types.go

@@ -35,6 +35,7 @@ type Config struct {
 	Threadiness          int
 	K3s                  K3sConfig
 
+	CustomizeSchemas func(context.Context, proxy.ClientGetter, *types.Schemas) error
 	GlobalSetup      func(context.Context) (context.Context, error)
 	MasterSetup      func(context.Context) (context.Context, error)
 	PreStart         func(context.Context) error