Misc improvements (#359)

* Support Minio

* Use Minio in quickstart

* Streamline quickstart

* Auto-ensure kafka topics

* Be nice throughout logs

.VERSION

@@ -1 +1 @@
-v0.11.7
+v0.11.8

.gitignore

@@ -9,3 +9,4 @@ honeypot-valid.json
 honeypot-invalid.json
 notes/*
 *.sql
+examples/quickstart/minio/*

.vscode/settings.json

@@ -1,5 +1,6 @@
 {
     "yaml.schemas": {
         "http://registry.silverton.io/io.silverton/honeypot/internal/config/app/v1.0.json": "/*conf*.yml"
+        // "./schemas/io.silverton/honeypot/internal/config/app/v1.0.json": "/*conf*.yml"
     }
 }

cmd/app.go

@@ -64,7 +64,7 @@ func (a *App) configure() {
 	if conf == "" {
 		conf = "config.yml"
 	}
-	log.Info().Msg("loading config from " + conf)
+	log.Info().Msg("🟢 loading config from " + conf)
 	viper.SetConfigFile(conf)
 	viper.SetConfigType("yaml")
 	err := viper.ReadInConfig()
@@ -84,21 +84,21 @@ func (a *App) configure() {
 }
 
 func (a *App) initializeStats() {
-	log.Info().Msg("initializing stats")
+	log.Info().Msg("🟢 initializing stats")
 	ps := stats.ProtocolStats{}
 	ps.Build()
 	a.stats = &ps
 }
 
 func (a *App) initializeSchemaCache() {
-	log.Info().Msg("initializing schema cache")
+	log.Info().Msg("🟢 initializing schema cache")
 	cache := cache.SchemaCache{}
 	cache.Initialize(a.config.SchemaCache)
 	a.schemaCache = &cache
 }
 
 func (a *App) initializeSinks() {
-	log.Info().Msg("initializing sinks")
+	log.Info().Msg("🟢 initializing sinks")
 	sinks, err := sink.BuildAndInitializeSinks(a.config.Sinks)
 	if err != nil {
 		log.Fatal().Err(err).Msg("could not build and init sinks")
@@ -107,7 +107,7 @@ func (a *App) initializeSinks() {
 }
 
 func (a *App) initializeManifold() {
-	log.Info().Msg("initializing manifold")
+	log.Info().Msg("🟢 initializing manifold")
 	manifold := manifold.SimpleManifold{}
 	err := manifold.Initialize(&a.sinks)
 	if err != nil {
@@ -117,61 +117,61 @@ func (a *App) initializeManifold() {
 }
 
 func (a *App) initializeRouter() {
-	log.Info().Msg("initializing router")
+	log.Info().Msg("🟢 initializing router")
 	a.engine = gin.New()
 	a.engine.SetTrustedProxies(nil)
 	a.engine.RedirectTrailingSlash = false
 }
 
 func (a *App) initializeMiddleware() {
-	log.Info().Msg("initializing middleware")
+	log.Info().Msg("🟢 initializing middleware")
 	a.engine.Use(gin.Recovery())
 	if a.config.Middleware.Timeout.Enabled {
-		log.Info().Msg("initializing request timeout middleware")
+		log.Info().Msg("🟢 initializing request timeout middleware")
 		a.engine.Use(middleware.Timeout(a.config.Middleware.Timeout))
 	}
 	if a.config.Middleware.RateLimiter.Enabled {
-		log.Info().Msg("initializing rate limiter middleware")
+		log.Info().Msg("🟢 initializing rate limiter middleware")
 		limiter := middleware.BuildRateLimiter(a.config.Middleware.RateLimiter)
 		limiterMiddleware := middleware.BuildRateLimiterMiddleware(limiter)
 		a.engine.Use(limiterMiddleware)
 	}
 	if a.config.Middleware.Cors.Enabled {
-		log.Info().Msg("initializing cors middleware")
+		log.Info().Msg("🟢 initializing cors middleware")
 		a.engine.Use(middleware.CORS(a.config.Middleware.Cors))
 	}
 	if a.config.Middleware.RequestLogger.Enabled {
-		log.Info().Msg("initializing request logger middleware")
+		log.Info().Msg("🟢 initializing request logger middleware")
 		a.engine.Use(middleware.RequestLogger())
 	}
 	if a.config.Middleware.Yeet.Enabled {
-		log.Info().Msg("initializing yeet middleware")
+		log.Info().Msg("🟢 initializing yeet middleware")
 		a.engine.Use(middleware.Yeet())
 	}
-	log.Info().Msg("initializing identity middleware")
+	log.Info().Msg("🟢 initializing identity middleware")
 	a.engine.Use(middleware.Identity(a.config.Identity))
 }
 
 func (a *App) initializeOpsRoutes() {
-	log.Info().Msg("initializing health check route")
+	log.Info().Msg("🟢 initializing health check route")
 	a.engine.GET(constants.HEALTH_PATH, handler.HealthcheckHandler)
-	log.Info().Msg("intializing stats route")
+	log.Info().Msg("🟢 intializing stats route")
 	a.engine.GET(constants.STATS_PATH, handler.StatsHandler(a.collectorMeta, a.stats))
-	log.Info().Msg("initializing overview routes")
+	log.Info().Msg("🟢 initializing overview routes")
 	a.engine.GET(constants.ROUTE_OVERVIEW_PATH, handler.RouteOverviewHandler(*a.config))
 	if a.config.App.EnableConfigRoute {
-		log.Info().Msg("initializing config overview")
+		log.Info().Msg("🟢 initializing config overview")
 		a.engine.GET(constants.CONFIG_OVERVIEW_PATH, handler.ConfigOverviewHandler(*a.config))
 	}
 }
 
 func (a *App) initializeSchemaCacheRoutes() {
 	if a.config.SchemaCache.Purge.Enabled {
-		log.Info().Msg("initializing schema cache purge route")
+		log.Info().Msg("🟢 initializing schema cache purge route")
 		a.engine.GET(a.config.SchemaCache.Purge.Path, handler.CachePurgeHandler(a.schemaCache))
 	}
 	if a.config.SchemaCache.SchemaDirectory.Enabled {
-		log.Info().Msg("initializing schema cache index and getter routes")
+		log.Info().Msg("🟢 initializing schema cache index and getter routes")
 		a.engine.GET(cache.SCHEMA_CACHE_ROOT_ROUTE, handler.CacheIndexHandler(a.schemaCache))
 		a.engine.GET(cache.SCHEMA_CACHE_ROOT_ROUTE+"/*"+cache.SCHEMA_ROUTE_PARAM, handler.CacheGetHandler(a.schemaCache))
 	}
@@ -180,21 +180,21 @@ func (a *App) initializeSchemaCacheRoutes() {
 func (a *App) initializeSnowplowRoutes() {
 	if a.config.Inputs.Snowplow.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing snowplow routes")
+		log.Info().Msg("🟢 initializing snowplow routes")
 		if a.config.Inputs.Snowplow.StandardRoutesEnabled {
-			log.Info().Msg("initializing standard snowplow routes")
+			log.Info().Msg("🟢 initializing standard snowplow routes")
 			a.engine.GET(constants.SNOWPLOW_STANDARD_GET_PATH, handler.SnowplowHandler(handlerParams))
 			a.engine.POST(constants.SNOWPLOW_STANDARD_POST_PATH, handler.SnowplowHandler(handlerParams))
 			if a.config.Inputs.Snowplow.OpenRedirectsEnabled {
-				log.Info().Msg("initializing standard open redirect route")
+				log.Info().Msg("🟢 initializing standard open redirect route")
 				a.engine.GET(constants.SNOWPLOW_STANDARD_REDIRECT_PATH, handler.SnowplowHandler(handlerParams))
 			}
 		}
-		log.Info().Msg("initializing custom snowplow routes")
+		log.Info().Msg("🟢 initializing custom snowplow routes")
 		a.engine.GET(a.config.Inputs.Snowplow.GetPath, handler.SnowplowHandler(handlerParams))
 		a.engine.POST(a.config.Inputs.Snowplow.PostPath, handler.SnowplowHandler(handlerParams))
 		if a.config.Inputs.Snowplow.OpenRedirectsEnabled {
-			log.Info().Msg("initializing custom open redirect route")
+			log.Info().Msg("🟢 initializing custom open redirect route")
 			a.engine.GET(a.config.Inputs.Snowplow.RedirectPath, handler.SnowplowHandler(handlerParams))
 		}
 	}
@@ -203,23 +203,23 @@ func (a *App) initializeSnowplowRoutes() {
 func (a *App) initializeGenericRoutes() {
 	if a.config.Inputs.Generic.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing generic routes")
+		log.Info().Msg("🟢 initializing generic routes")
 		a.engine.POST(a.config.Inputs.Generic.Path, handler.GenericHandler(handlerParams))
 	}
 }
 
 func (a *App) initializeCloudeventsRoutes() {
 	if a.config.Inputs.Cloudevents.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing cloudevents routes")
+		log.Info().Msg("🟢 initializing cloudevents routes")
 		a.engine.POST(a.config.Inputs.Cloudevents.Path, handler.CloudeventsHandler(handlerParams))
 	}
 }
 
 func (a *App) initializeWebhookRoutes() {
 	if a.config.Inputs.Webhook.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing webhook routes")
+		log.Info().Msg("🟢 initializing webhook routes")
 		a.engine.POST(a.config.Inputs.Webhook.Path, handler.WebhookHandler(handlerParams))
 		a.engine.POST(a.config.Inputs.Webhook.Path+"/*"+constants.HONEYPOT_SCHEMA_PARAM, handler.WebhookHandler(handlerParams))
 	}
@@ -228,7 +228,7 @@ func (a *App) initializeWebhookRoutes() {
 func (a *App) initializePixelRoutes() {
 	if a.config.Inputs.Pixel.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing pixel routes")
+		log.Info().Msg("🟢 initializing pixel routes")
 		a.engine.GET(a.config.Inputs.Pixel.Path, handler.PixelHandler(handlerParams))
 		a.engine.GET(a.config.Inputs.Pixel.Path+"/*"+constants.HONEYPOT_SCHEMA_PARAM, handler.PixelHandler(handlerParams))
 	}
@@ -237,7 +237,7 @@ func (a *App) initializePixelRoutes() {
 func (a *App) initializeSquawkboxRoutes() {
 	if a.config.Squawkbox.Enabled {
 		handlerParams := a.handlerParams()
-		log.Info().Msg("initializing squawkbox routes")
+		log.Info().Msg("🟢 initializing squawkbox routes")
 		a.engine.POST(a.config.Squawkbox.CloudeventsPath, handler.SquawkboxHandler(handlerParams, protocol.CLOUDEVENTS))
 		a.engine.POST(a.config.Squawkbox.GenericPath, handler.SquawkboxHandler(handlerParams, protocol.GENERIC))
 		a.engine.POST(a.config.Squawkbox.SnowplowPath, handler.SquawkboxHandler(handlerParams, protocol.SNOWPLOW))
@@ -246,7 +246,7 @@ func (a *App) initializeSquawkboxRoutes() {
 }
 
 func (a *App) Initialize() {
-	log.Info().Msg("initializing app")
+	log.Info().Msg("🟢 initializing app")
 	a.configure()
 	a.initializeStats()
 	a.initializeSinks()
@@ -273,14 +273,14 @@ func (a *App) Run() {
 	}
 	go func() {
 		if err := srv.ListenAndServe(); err != nil && errors.Is(err, http.ErrServerClosed) {
-			log.Info().Msgf("server shut down")
+			log.Info().Msgf("🟢 server shut down")
 		}
 	}()
 	// Safe shutdown
 	quit := make(chan os.Signal, 1)
 	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
 	<-quit
-	log.Info().Msg("shutting down server...")
+	log.Info().Msg("🟢 shutting down server...")
 	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
 	if err := srv.Shutdown(ctx); err != nil {

examples/quickstart/docker-compose.yml

@@ -3,7 +3,7 @@ version: "3.9"
 services:
   honeypot:
     container_name: honeypot
-    image: ghcr.io/silverton-io/honeypot:v0.11.7
+    image: ghcr.io/silverton-io/honeypot:v0.11.8
     volumes:
       - type: bind
         source: ./honeypot/quickstart.conf.yml
@@ -22,17 +22,36 @@ services:
       test: curl -f localhost:8080/health
       interval: 15s
     depends_on:
-      redpanda-init:
-        condition: service_completed_successfully
+      redpanda-1:
+        condition: service_healthy
+      redpanda-2:
+        condition: service_healthy
+      redpanda-3:
+        condition: service_healthy
 
   sample-ui:
     container_name: sample-ui
     image: ghcr.io/silverton-io/sample-tracked-ui:0.1.1
     ports:
       - 8081:8080
 
+  minio:
+    container_name: minio
+    image: quay.io/minio/minio
+    command: server /data --console-address ":9001"
+    ports:
+      - 9000:9000
+      - 9001:9001
+    volumes:
+      - ./minio/:/data
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
+
   redpanda-1:
-    image: docker.vectorized.io/vectorized/redpanda:v21.11.15 # Note! Use the latest version :)
+    image: docker.vectorized.io/vectorized/redpanda:latest # Note! Use the latest version :)
     container_name: redpanda-1
     command:
       - redpanda start
@@ -58,7 +77,7 @@ services:
       start_period: 30s
 
   redpanda-2:
-    image: docker.vectorized.io/vectorized/redpanda:v21.11.15 # Note! Use the latest version :)
+    image: docker.vectorized.io/vectorized/redpanda:latest # Note! Use the latest version :)
     container_name: redpanda-2
     command:
       - redpanda start
@@ -85,7 +104,7 @@ services:
       start_period: 30s
 
   redpanda-3:
-    image: docker.vectorized.io/vectorized/redpanda:v21.11.15 # Note! Use the latest version :)
+    image: docker.vectorized.io/vectorized/redpanda:latest # Note! Use the latest version :)
     container_name: redpanda-3
     command:
       - redpanda start
@@ -111,30 +130,6 @@ services:
       interval: 1s
       start_period: 30s
 
-  redpanda-init:
-    image: docker.vectorized.io/vectorized/redpanda:v21.11.15 # Note! Use the latest version :)
-    container_name: redpanda-init
-    command:
-      - cluster metadata --brokers redpanda-1:29092,redpanda-2:29093,redpanda-3:29094
-    depends_on:
-      redpanda-1:
-        condition: service_healthy
-      redpanda-2:
-        condition: service_healthy
-      redpanda-3:
-        condition: service_healthy
-    deploy:
-      restart_policy:
-        condition: on-failure
-
-  materialized:
-    container_name: materialized
-    image: materialize/materialized # Note! Use the latest version :)
-    init: true
-    command: -w2 --disable-telemetry
-    ports:
-      - 6875:6875
-
   kowl:
     image: quay.io/cloudhut/kowl:v1.5.0
     restart: on-failure
@@ -152,6 +147,20 @@ services:
       redpanda-3:
         condition: service_healthy
 
+  materialize:
+    container_name: materialize
+    image: materialize/materialized # Note! Use the latest version :)
+    init: true
+    command: -w2 --disable-telemetry
+    ports:
+      - 6875:6875
+
   mzcli:
     container_name: mzcli
     image: materialize/cli
+    volumes:
+      - ./materialize/:/cmd
+    command: -h materialize -p 6875 -U materialize -f /cmd/create_sources_and_views.sql
+    depends_on:
+      honeypot:
+        condition: service_healthy

examples/quickstart/honeypot/quickstart.conf.yml

@@ -74,8 +74,12 @@ inputs:
 
 schemaCache:
   backend:
-    type: fs
-    path: /schemas
+    type: minio
+    minioEndpoint: minio:9000
+    accessKeyId: minioadmin
+    secretAccessKey: minioadmin
+    bucket: honeypot-schemas
+    path: /
   ttlSeconds: 300
   maxSizeBytes: 104857600 # 100mb -> 100 * 1024 * 1024
   purge:

examples/quickstart/minio/.minio.sys/format.json

@@ -0,0 +1 @@
+{"version":"1","format":"xl-single","id":"28c9bb40-1ee3-49e8-9b74-0db7b34f9f83","xl":{"version":"3","this":"d25ecab2-abe0-4fb4-915a-fe804a06dd8e","sets":[["d25ecab2-abe0-4fb4-915a-fe804a06dd8e"]],"distributionAlgo":"SIPMOD+PARITY"}}