Move to generic types

Author: howardjohn

adsc/adsc.go

@@ -24,8 +24,8 @@ import (
 	structpb "github.com/golang/protobuf/ptypes/struct"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
+	"istio.io/istio/pkg/log"
 	"istio.io/istio/pkg/util/sets"
-	"istio.io/pkg/log"
 )
 
 var (

adsc/load.go

@@ -8,11 +8,7 @@ import (
 )
 
 func Fetch(pilotAddress string, config *Config) (*Responses, error) {
-	log := func(template string, args ...interface{}) {
-		a := []interface{}{"%v: " + template, config.Workload}
-		a = append(a, args...)
-		scope.Infof(a...)
-	}
+	log := scope.WithLabels("workload", config.Workload).Infof
 	watchAll := map[string]struct{}{"cds": {}, "eds": {}, "rds": {}, "lds": {}}
 	log("Connecting: %v", config.IP)
 	con, err := Dial(pilotAddress, config)
@@ -51,11 +47,7 @@ func Fetch(pilotAddress string, config *Config) (*Responses, error) {
 
 func Connect(pilotAddress string, config *Config) {
 	attempts := 0
-	log := func(template string, args ...interface{}) {
-		a := []interface{}{"%v: " + template, config.Workload}
-		a = append(a, args...)
-		scope.Infof(a...)
-	}
+	log := scope.WithLabels("workload", config.Workload).Infof
 	// Follow envoy defaults https://github.com/envoyproxy/envoy/blob/v1.12.1/source/common/config/grpc_stream.h#L40-L43
 	b := backoff.NewExponentialBackOff()
 	b.MaxElapsedTime = 0

cmd/cluster.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/ghodss/yaml"
 	"github.com/spf13/cobra"
-	"istio.io/pkg/log"
+	"istio.io/istio/pkg/log"
 
 	"github.com/howardjohn/pilot-load/pkg/simulation"
 	"github.com/howardjohn/pilot-load/pkg/simulation/model"

cmd/cmd.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/ghodss/yaml"
 	"github.com/spf13/cobra"
-	"istio.io/pkg/log"
+	"istio.io/istio/pkg/log"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
@@ -107,7 +107,7 @@ func setDefaultArgs(args model.Args) (model.Args, error) {
 		return model.Args{}, err
 	}
 	if _, f := xdsMetadata[CLOUDRUN_ADDR]; !f && args.Auth.Type == security.AuthTypeGoogle {
-		mwh, err := args.Client.Kubernetes.AdmissionregistrationV1().MutatingWebhookConfigurations().Get(context.Background(), "istiod-asm-managed", metav1.GetOptions{})
+		mwh, err := args.Client.Kube().AdmissionregistrationV1().MutatingWebhookConfigurations().Get(context.Background(), "istiod-asm-managed", metav1.GetOptions{})
 		if err != nil {
 			return model.Args{}, fmt.Errorf("failed to default CLOUDRUN_ADDR: %v", err)
 		}
@@ -130,8 +130,6 @@ var rootCmd = &cobra.Command{
 	Short:        "open XDS connections to pilot",
 	SilenceUsage: true,
 	PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
-		log.FindScope("dump").SetOutputLevel(log.WarnLevel)
-
 		return log.Configure(loggingOptions)
 	},
 }
@@ -154,6 +152,7 @@ func init() {
 		xdsLatencyCmd,
 		reproduceCmd,
 		dumpCmd,
+		isolatedCmd,
 	)
 }
 

cmd/impersonate.go

@@ -16,7 +16,6 @@ func init() {
 	impersonateCmd.PersistentFlags().DurationVar(&impersonateConfig.Delay, "delay", impersonateConfig.Delay, "delay between each connection")
 	impersonateCmd.PersistentFlags().IntVar(&impersonateConfig.Replicas, "replicas", impersonateConfig.Replicas, "number of connections to make for each pod")
 	impersonateCmd.PersistentFlags().StringVar(&impersonateConfig.Selector, "selector", impersonateConfig.Selector, "selector to use {sidecar,external,both}")
-	impersonateCmd.PersistentFlags().BoolVar(&impersonateConfig.Watch, "watch", impersonateConfig.Watch, "watch for pod changes")
 }
 
 var impersonateCmd = &cobra.Command{

cmd/isolated.go

@@ -0,0 +1,128 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"runtime/pprof"
+
+	"github.com/spf13/cobra"
+	"istio.io/istio/pilot/pkg/features"
+	"istio.io/istio/pilot/pkg/xds"
+	"istio.io/istio/pkg/log"
+	"istio.io/istio/pkg/test"
+	"k8s.io/apimachinery/pkg/watch"
+
+	"github.com/howardjohn/pilot-load/pkg/kube"
+	"github.com/howardjohn/pilot-load/pkg/simulation"
+	"github.com/howardjohn/pilot-load/pkg/simulation/isolated"
+	"github.com/howardjohn/pilot-load/pkg/simulation/model"
+	"github.com/howardjohn/pilot-load/pkg/simulation/monitoring"
+	"github.com/howardjohn/pilot-load/pkg/simulation/security"
+)
+
+func init() {
+	isolatedCmd.PersistentFlags().StringVarP(&configFile, "config", "c", configFile, "config file")
+}
+
+func WithProfiling(c *cobra.Command) *cobra.Command {
+	var cpuProfile string
+	var memProfile string
+	c.PersistentFlags().StringVar(&cpuProfile, "cpuprofile", cpuProfile, "file to write cpu profile to")
+	c.PersistentFlags().StringVar(&memProfile, "memprofile", memProfile, "file to write memory profile to")
+	orig := c.RunE
+	c.RunE = func(cmd *cobra.Command, args []string) error {
+		if cpuProfile != "" {
+			f, err := os.Create(cpuProfile)
+			if err != nil {
+				return err
+			}
+			if err := pprof.StartCPUProfile(f); err != nil {
+				return err
+			}
+			defer pprof.StopCPUProfile()
+		}
+		if memProfile != "" {
+			f, err := os.Create(memProfile)
+			if err != nil {
+				return err
+			}
+			defer func() {
+				pprof.WriteHeapProfile(f)
+			}()
+		}
+		return orig(cmd, args)
+	}
+	return c
+}
+
+func executeSimulations(a model.Args, s model.Simulation) error {
+	ctx, cancel := context.WithCancel(context.Background())
+	go simulation.CaptureTermination(ctx, cancel)
+	defer cancel()
+	simulationContext := model.Context{Context: ctx, Args: a, Client: a.Client, Cancel: cancel}
+	if err := s.Run(simulationContext); err != nil {
+		log.Errorf("failed: %v, starting cleanup", err)
+		cleanupErr := s.Cleanup(simulationContext)
+		return fmt.Errorf("failed to run: %v; cleanup: %v", err, cleanupErr)
+	}
+	<-ctx.Done()
+	return s.Cleanup(simulationContext)
+}
+
+var isolatedCmd = WithProfiling(&cobra.Command{
+	Use:   "isolated",
+	Short: "simulate a full cluster in a single binary",
+	RunE: func(cmd *cobra.Command, _ []string) error {
+		var ds *xds.FakeDiscoveryServer
+		ready := make(chan struct{})
+		done := make(chan struct{})
+		defer func() {
+			close(done)
+		}()
+		// Bump up QPS of requests so test starts faster
+		features.RequestLimit = 200.0
+		// Kube fake explodes too early
+		watch.DefaultChanSize = 10_000
+		go test.Wrap(func(t test.Failer) {
+			ds = xds.NewFakeDiscoveryServer(t, xds.FakeOptions{
+				ListenerBuilder: func() (net.Listener, error) {
+					return net.Listen("tcp", "127.0.0.1:0")
+				},
+			})
+			close(ready)
+			<-done
+		})
+		<-ready
+		ms := monitoring.StartMonitoring(8765)
+		ds.Discovery.InitDebug(ms.Handler.(*http.ServeMux), false, func() map[string]string {
+			return nil
+		})
+
+		args := model.Args{
+			PilotAddress: ds.Listener.Addr().String(),
+			Client:       kube.NewFakeClient(ds.KubeClient()),
+			Auth: &security.AuthOptions{
+				Type: security.AuthTypePlaintext,
+			},
+		}
+		config, err := readConfigFile(configFile)
+		if err != nil {
+			return fmt.Errorf("failed to read config file: %v", err)
+		}
+		config = config.ApplyDefaults()
+
+		logConfig(config)
+		logClusterConfig(config)
+		log.Infof("Starting cluster, total size: %v pods", config.PodCount())
+
+		sim := isolated.NewCluster(isolated.IsolatedSpec{Config: config})
+		if err := executeSimulations(args, sim); err != nil {
+			return fmt.Errorf("error executing: %v", err)
+		}
+
+		return nil
+	},
+})

examples/big.yaml

@@ -1,14 +1,14 @@
 nodeMetadata: {}
 jitter:
-  workloads: "30s"
+  workloads: "110ms"
   config: "0s"
 namespaces:
 - name: mesh
   replicas: 1
   applications:
   - name: big
-    replicas: 1
-    instances: 1000
+    replicas: 200
+    instances: 10
 nodes:
 - name: node
   count: 10