Add telemetry to config (#152)

* Demo Mode MVP

* messages improve

* downloading based on the OS

* downloading based on the OS

* downloading based on the OS

* Modeler keep running

* A lot of revisions comes now

* Fix color

agent/main.go

@@ -4,12 +4,6 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
-	"github.com/gin-contrib/static"
-	"github.com/gin-gonic/gin"
-	"github.com/gorilla/websocket"
-	"github.com/romana/rlog"
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/tap"
 	"mizuserver/pkg/api"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/routes"
@@ -19,9 +13,17 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+
+	"github.com/gin-contrib/static"
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/romana/rlog"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/tap"
 )
 
 var shouldTap = flag.Bool("tap", false, "Run in tapper mode without API")
+var demo = flag.Bool("demo", false, "Run in Demo mode with API")
 var apiServer = flag.Bool("api-server", false, "Run in API server mode with API")
 var standalone = flag.Bool("standalone", false, "Run in standalone tapper and API mode")
 var apiServerAddress = flag.String("api-server-address", "", "Address of mizu API server")
@@ -34,13 +36,12 @@ func main() {
 	if !*shouldTap && !*apiServer && !*standalone {
 		panic("One of the flags --tap, --api or --standalone must be provided")
 	}
-
 	if *standalone {
 		harOutputChannel, outboundLinkOutputChannel := tap.StartPassiveTapper(tapOpts)
 		filteredHarChannel := make(chan *tap.OutputChannelItem)
 
 		go filterHarItems(harOutputChannel, filteredHarChannel, getTrafficFilteringOptions())
-		go api.StartReadingEntries(filteredHarChannel, nil)
+		go api.StartReadingEntries(filteredHarChannel, nil, false)
 		go api.StartReadingOutbound(outboundLinkOutputChannel)
 
 		hostApi(nil)
@@ -69,7 +70,12 @@ func main() {
 		filteredHarChannel := make(chan *tap.OutputChannelItem)
 
 		go filterHarItems(socketHarOutChannel, filteredHarChannel, getTrafficFilteringOptions())
-		go api.StartReadingEntries(filteredHarChannel, nil)
+		if *demo {
+			workdir := "./hars"
+			go api.StartReadingEntries(filteredHarChannel, &workdir, true)
+		} else {
+			go api.StartReadingEntries(filteredHarChannel, nil, false)
+		}
 
 		hostApi(socketHarOutChannel)
 	}

agent/pkg/api/main.go

@@ -5,10 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/google/martian/har"
-	"github.com/romana/rlog"
-	"github.com/up9inc/mizu/tap"
-	"go.mongodb.org/mongo-driver/bson/primitive"
+	"math/rand"
 	"mizuserver/pkg/holder"
 	"net/url"
 	"os"
@@ -17,6 +14,11 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/martian/har"
+	"github.com/romana/rlog"
+	"github.com/up9inc/mizu/tap"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/resolver"
@@ -47,15 +49,15 @@ func init() {
 	holder.SetResolver(res)
 }
 
-func StartReadingEntries(harChannel <-chan *tap.OutputChannelItem, workingDir *string) {
+func StartReadingEntries(harChannel <-chan *tap.OutputChannelItem, workingDir *string, demo bool) {
 	if workingDir != nil && *workingDir != "" {
-		startReadingFiles(*workingDir)
+		startReadingFiles(*workingDir, demo)
 	} else {
 		startReadingChannel(harChannel)
 	}
 }
 
-func startReadingFiles(workingDir string) {
+func startReadingFiles(workingDir string, infiniteLoad bool) {
 	err := os.MkdirAll(workingDir, os.ModePerm)
 	utils.CheckErr(err)
 
@@ -86,18 +88,23 @@ func startReadingFiles(workingDir string) {
 		utils.CheckErr(decErr)
 
 		for _, entry := range inputHar.Log.Entries {
-			time.Sleep(time.Millisecond * 250)
+			if infiniteLoad {
+				entry.StartedDateTime = time.Now().Add(20 * time.Millisecond)
+			}
+			time.Sleep(time.Millisecond * time.Duration(rand.Intn(300)))
 			connectionInfo := &tap.ConnectionInfo{
-				ClientIP: fileInfo.Name(),
+				ClientIP:   fileInfo.Name(),
 				ClientPort: "",
-				ServerIP: "",
+				ServerIP:   "",
 				ServerPort: "",
 				IsOutgoing: false,
 			}
 			saveHarToDb(entry, connectionInfo)
 		}
-		rmErr := os.Remove(inputFilePath)
-		utils.CheckErr(rmErr)
+		if !infiniteLoad {
+			rmErr := os.Remove(inputFilePath)
+			utils.CheckErr(rmErr)
+		}
 	}
 }
 
@@ -118,7 +125,6 @@ func StartReadingOutbound(outboundLinkChannel <-chan *tap.OutboundLink) {
 	}
 }
 
-
 func saveHarToDb(entry *har.Entry, connectionInfo *tap.ConnectionInfo) {
 	entryBytes, _ := json.Marshal(entry)
 	serviceName, urlPath := getServiceNameFromUrl(entry.Request.URL)
@@ -196,6 +202,5 @@ func getEstimatedEntrySizeBytes(mizuEntry models.MizuEntry) int {
 	sizeBytes += 8 // SizeBytes bytes
 	sizeBytes += 1 // IsOutgoing bytes
 
-
 	return sizeBytes
 }

cli/cmd/demo.go

@@ -0,0 +1,32 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+type MizuDemoOptions struct {
+	GuiPort            uint16
+	Analyze            bool
+	AnalyzeDestination string
+}
+
+var mizuDemoOptions = &MizuDemoOptions{}
+
+var demoCmd = &cobra.Command{
+	Use:   "demo",
+	Short: "Record ingoing traffic of a kubernetes pod",
+	Long: `Record the ingoing traffic of a kubernetes pod.
+Supported protocols are HTTP and gRPC.`,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		RunMizuTapDemo(mizuDemoOptions)
+		return nil
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(demoCmd)
+
+	demoCmd.Flags().Uint16VarP(&mizuDemoOptions.GuiPort, "gui-port", "p", 8899, "Provide a custom port for the web interface webserver")
+	demoCmd.Flags().BoolVar(&mizuDemoOptions.Analyze, "analyze", false, "Uploads traffic to UP9 cloud for further analysis (Beta)")
+	demoCmd.Flags().StringVar(&mizuDemoOptions.AnalyzeDestination, "dest", "up9.app", "Destination environment")
+}

cli/cmd/demoRunner.go

@@ -0,0 +1,185 @@
+package cmd
+
+import (
+	"archive/zip"
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/up9inc/mizu/cli/uiUtils"
+)
+
+func RunMizuTapDemo(demoOptions *MizuDemoOptions) {
+	dir, _ := os.Getwd()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	downloadMizuDemo(dir)
+
+	go callMizuDemo(ctx, cancel, dir, demoOptions)
+	if demoOptions.Analyze {
+		go analyze(demoOptions)
+		fmt.Printf(uiUtils.Purple, "mizu tap \"catalogue-.*|carts-[0-9].*|payment.*|shipping.*|user-[0-9].*\" -n sock-shop --analyze\n")
+	} else {
+		fmt.Printf(uiUtils.Purple, "mizu tap \"catalogue-.*|carts-[0-9].*|payment.*|shipping.*|user-[0-9].*\" -n sock-shop\n")
+	}
+	fmt.Println("Mizu will be available on http://localhost:8899 in a few seconds")
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+
+	select {
+	case <-ctx.Done():
+		break
+	case <-sigChan:
+		cleanUpDemoResources(dir)
+		cancel()
+	}
+}
+
+func cleanUpDemoResources(dir string) {
+	removeFile(fmt.Sprintf("%s/site.zip", dir))
+	removeFile(fmt.Sprintf("%s/site", dir))
+	removeFile(fmt.Sprintf("%s/apiserver.zip", dir))
+	removeFile(fmt.Sprintf("%s/apiserver", dir))
+	removeFile(fmt.Sprintf("%s/entries.db", dir))
+	removeFile(fmt.Sprintf("%s/hars", dir))
+	removeFile(fmt.Sprintf("%s/hars.zip", dir))
+}
+
+func removeFile(file string) {
+	err := os.RemoveAll(file)
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func downloadMizuDemo(dir string) {
+	if runtime.GOOS != "darwin" && runtime.GOOS != "linux" {
+		panic("Platform not supported")
+	}
+	mizuApiURL := fmt.Sprintf("https://storage.googleapis.com/up9-mizu-demo-mode/apiserver-%s.zip", runtime.GOOS)
+	siteFileURL := "https://storage.googleapis.com/up9-mizu-demo-mode/site.zip"
+	harsURL := "https://storage.googleapis.com/up9-mizu-demo-mode/hars.zip"
+
+	dirApi := fmt.Sprintf("%s/apiserver.zip", dir)
+	dirSite := fmt.Sprintf("%s/site.zip", dir)
+	dirHars := fmt.Sprintf("%s/hars.zip", dir)
+
+	DownloadFile(dirApi, mizuApiURL)
+	DownloadFile(dirSite, siteFileURL)
+	DownloadFile(dirHars, harsURL)
+
+	UnzipSite(dirSite, fmt.Sprintf("%s/", dir))
+	UnzipSite(dirApi, fmt.Sprintf("%s/", dir))
+	UnzipSite(dirHars, fmt.Sprintf("%s/", dir))
+	allowExecutable(fmt.Sprintf("%s/apiserver", dir))
+}
+
+func DownloadFile(filepath string, url string) error {
+
+	resp, err := http.Get(url)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	out, err := os.Create(filepath)
+	if err != nil {
+		return err
+	}
+	defer out.Close()
+	_, err = io.Copy(out, resp.Body)
+	return err
+}
+
+func UnzipSite(src string, dest string) ([]string, error) {
+	var filenames []string
+
+	r, err := zip.OpenReader(src)
+	if err != nil {
+		return filenames, err
+	}
+	defer r.Close()
+
+	for _, f := range r.File {
+
+		fpath := filepath.Join(dest, f.Name)
+		if !strings.HasPrefix(fpath, filepath.Clean(dest)+string(os.PathSeparator)) {
+			return filenames, fmt.Errorf("%s: illegal file path", fpath)
+		}
+
+		filenames = append(filenames, fpath)
+
+		if f.FileInfo().IsDir() {
+			os.MkdirAll(fpath, os.ModePerm)
+			continue
+		}
+		if err = os.MkdirAll(filepath.Dir(fpath), os.ModePerm); err != nil {
+			return filenames, err
+		}
+
+		outFile, err := os.OpenFile(fpath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, f.Mode())
+		if err != nil {
+			return filenames, err
+		}
+
+		rc, err := f.Open()
+		if err != nil {
+			return filenames, err
+		}
+
+		_, err = io.Copy(outFile, rc)
+		outFile.Close()
+		rc.Close()
+
+		if err != nil {
+			return filenames, err
+		}
+	}
+	return filenames, nil
+}
+
+func allowExecutable(dir string) {
+	if err := os.Chmod(dir, 0755); err != nil {
+		log.Fatalln(err)
+	}
+}
+
+func callMizuDemo(ctx context.Context, cancel context.CancelFunc, dir string, demoOptions *MizuDemoOptions) {
+	cmd := exec.Command(fmt.Sprintf("%s/apiserver", dir), "--aggregator", "--demo")
+	var out bytes.Buffer
+
+	// set the output to our variable
+	cmd.Stdout = &out
+	err := cmd.Run()
+	if err != nil {
+		log.Println(err)
+	}
+}
+
+func analyze(demoOptions *MizuDemoOptions) {
+	mizuProxiedUrl := getMizuCollectorProxiedHostAndPath(demoOptions.GuiPort)
+	for {
+		response, err := http.Get(fmt.Sprintf("http://%s/api/uploadEntries?dest=%s&interval=10", mizuProxiedUrl, demoOptions.AnalyzeDestination))
+		if err != nil || response.StatusCode != 200 {
+			fmt.Printf(uiUtils.Red, "Mizu Not running, waiting 10 seconds before trying again\n")
+		} else {
+			fmt.Printf(uiUtils.Purple, "Traffic is uploading to UP9 cloud for further analsys\n")
+			break
+		}
+		time.Sleep(10 * time.Second)
+	}
+}
+
+func getMizuCollectorProxiedHostAndPath(mizuPort uint16) string {
+	return fmt.Sprintf("localhost:%d", mizuPort)
+}

cli/mizu/config.go

@@ -21,13 +21,13 @@ type CommandLineFlag struct {
 	CommandLineName   string
 	YamlHierarchyName string
 	DefaultValue      interface{}
-	Type              reflect.Kind
 }
 
 const (
 	ConfigurationKeyAnalyzingDestination = "tap.dest"
 	ConfigurationKeyUploadInterval       = "tap.uploadInterval"
 	ConfigurationKeyMizuImage            = "mizuImage"
+	ConfigurationKeyTelemetry            = "telemetry"
 )
 
 var allowedSetFlags = []CommandLineFlag{
@@ -35,20 +35,22 @@ var allowedSetFlags = []CommandLineFlag{
 		CommandLineName:   "dest",
 		YamlHierarchyName: ConfigurationKeyAnalyzingDestination,
 		DefaultValue:      "up9.app",
-		Type:              reflect.String,
 		// TODO: maybe add short description that we can show
 	},
 	{
 		CommandLineName:   "uploadInterval",
 		YamlHierarchyName: ConfigurationKeyUploadInterval,
 		DefaultValue:      10,
-		Type:              reflect.Int,
 	},
 	{
 		CommandLineName:   "mizuImage",
 		YamlHierarchyName: ConfigurationKeyMizuImage,
 		DefaultValue:      fmt.Sprintf("gcr.io/up9-docker-hub/mizu/%s:%s", Branch, SemVer),
-		Type:              reflect.String,
+	},
+	{
+		CommandLineName:   "telemetry",
+		YamlHierarchyName: ConfigurationKeyTelemetry,
+		DefaultValue:      true,
 	},
 }
 
@@ -56,13 +58,25 @@ func GetString(key string) string {
 	return fmt.Sprintf("%v", getValueFromMergedConfig(key))
 }
 
+func GetBool(key string) bool {
+	stringVal := GetString(key)
+	Log.Debugf("Found string value %v", stringVal)
+
+	val, err := strconv.ParseBool(stringVal)
+	if err != nil {
+		Log.Warningf(uiUtils.Red, fmt.Sprintf( "Invalid value %v for key %s, expected bool", stringVal, key))
+		os.Exit(1)
+	}
+	return val
+}
+
 func GetInt(key string) int {
 	stringVal := GetString(key)
 	Log.Debugf("Found string value %v", stringVal)
 
 	val, err := strconv.Atoi(stringVal)
 	if err != nil {
-		Log.Warningf("Invalid value %v for key %s", val, key)
+		Log.Warningf(uiUtils.Red, fmt.Sprintf("Invalid value %v for key %s, expected int", stringVal, key))
 		os.Exit(1)
 	}
 	return val
@@ -140,13 +154,13 @@ func mergeConfigFile() error {
 
 func addToConfig(prefix string, value interface{}) {
 	typ := reflect.TypeOf(value).Kind()
-	if typ == reflect.Int || typ == reflect.String || typ == reflect.Slice {
-		validateConfigFileKey(prefix)
-		configObj[prefix] = value
-	} else if typ == reflect.Map {
+	if typ == reflect.Map {
 		for k1, v1 := range value.(map[string]interface{}) {
 			addToConfig(fmt.Sprintf("%s.%s", prefix, k1), v1)
 		}
+	} else {
+		validateConfigFileKey(prefix)
+		configObj[prefix] = value
 	}
 }
 
@@ -161,26 +175,23 @@ func mergeCommandLineFlags(commandLineValues []string) error {
 			return errors.New(fmt.Sprintf("invalid set argument %s", e))
 		}
 		setFlagKey, argumentValue := split[0], split[1]
-		argumentNameInConfig, expectedType, err := flagFromAllowed(setFlagKey)
+		argumentNameInConfig, err := flagFromAllowed(setFlagKey)
 		if err != nil {
 			return err
 		}
-		argumentType := reflect.ValueOf(argumentValue).Kind()
-		if argumentType != expectedType {
-			return errors.New(fmt.Sprintf("Invalid value for argument %s (should be type %s but got %s", setFlagKey, expectedType, argumentType))
-		}
+
 		configObj[argumentNameInConfig] = argumentValue
 	}
 	return nil
 }
 
-func flagFromAllowed(setFlagKey string) (string, reflect.Kind, error) {
+func flagFromAllowed(setFlagKey string) (string, error) {
 	for _, allowedFlag := range allowedSetFlags {
 		if strings.ToLower(allowedFlag.CommandLineName) == strings.ToLower(setFlagKey) {
-			return allowedFlag.YamlHierarchyName, allowedFlag.Type, nil
+			return allowedFlag.YamlHierarchyName, nil
 		}
 	}
-	return "", reflect.Invalid, errors.New(fmt.Sprintf("invalid set argument %s", setFlagKey))
+	return "", errors.New(fmt.Sprintf("invalid set argument %s", setFlagKey))
 }
 
 func validateConfigFileKey(configFileKey string) {
@@ -195,7 +206,7 @@ func validateConfigFileKey(configFileKey string) {
 
 func addToConfigObj(key string, value interface{}, configObj map[string]interface{}) {
 	typ := reflect.TypeOf(value).Kind()
-	if typ == reflect.Int || typ == reflect.String || typ == reflect.Slice {
+	if typ != reflect.Map {
 		if strings.Contains(key, ".") {
 			split := strings.SplitN(key, ".", 2)
 			firstLevelKey := split[0]
@@ -208,4 +219,3 @@ func addToConfigObj(key string, value interface{}, configObj map[string]interfac
 		}
 	}
 }
-

cli/mizu/telemetry.go

@@ -10,6 +10,11 @@ import (
 const telemetryUrl = "https://us-east4-up9-prod.cloudfunctions.net/mizu-telemetry"
 
 func ReportRun(cmd string, args interface{}) {
+	if !GetBool(ConfigurationKeyTelemetry) {
+		Log.Debugf("not reporting due to config value")
+		return
+	}
+
 	if Branch != "main" {
 		Log.Debugf("reporting only on main branch")
 		return