flowd: an opinionated daemon for collecting system telemetry

.github/release.yml

@@ -0,0 +1,11 @@
+changelog:
+  categories:
+    - title: 🌱 What's new!
+      labels:
+        - feature
+    - title: 🐆 Optimizations
+      labels:
+        - performance
+    - title: 🛠️ Bugfixes
+      labels:
+        - bug

.github/workflows/ci.yml

@@ -0,0 +1,42 @@
+# This workflow will build a golang project
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go
+
+name: flowd
+
+on:
+  push:
+    branches: [ main ]
+    tags:
+      - '*'
+  pull_request:
+    branches: [ main ]
+
+jobs:
+
+  build:
+    runs-on:
+      labels: OSS-Runner
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Set up Go
+      uses: actions/setup-go@v3
+      with:
+        go-version: "1.20"
+        cache: true
+
+    - name: Start containers
+      run: docker-compose -f "docker-compose.yml" up -d
+
+    - name: Test
+      run: make test
+
+    - name: Fmtcheck
+      run: make fmtcheck
+
+    - name: Lint
+      run: make lint
+
+    - name: Stop containers
+      run: docker-compose -f "docker-compose.yml" down -v
+      if: always()

.github/workflows/release.yml

@@ -0,0 +1,34 @@
+name: release
+
+on:
+  push:
+    # run only against tags
+    tags:
+      - '*'
+
+permissions:
+  contents: write
+  # packages: write
+  # issues: write
+
+jobs:
+  public:
+    runs-on: macos-latest
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 0
+    - run: git fetch --force --tags
+    - uses: actions/setup-go@v4
+      with:
+        go-version: stable
+      # More assembly might be required: Docker logins, GPG, etc. It all depends
+      # on your needs.
+    - uses: goreleaser/goreleaser-action@v4
+      with:
+        # either 'goreleaser' (default) or 'goreleaser-pro':
+        distribution: goreleaser
+        version: latest
+        args: release --debug --clean -f .goreleaser.public.yaml
+      env:
+        GITHUB_TOKEN: ${{ secrets.GO_RELEASER }}

.gitignore

@@ -1 +1,2 @@
 .coverprofile
+dist

.goreleaser.public.yaml

@@ -0,0 +1,58 @@
+# ⚠️  Find delightfully thorough documentation at https://goreleaser.com
+
+before:
+  hooks:
+    # You may remove this if you don't use go modules.
+    - go mod tidy
+    # you may remove this if you don't need go generate
+    # - go generate ./...
+builds:
+  - main: ./cmd/flowd
+    id: flowd
+    binary: flowd
+    goos:
+      - linux
+      - darwin
+      - windows
+
+archives:
+  - format: tar.gz
+    id: flowd
+    # this name template makes the OS and Arch compatible with the results of uname.
+    name_template: >-
+      flowd-
+      {{- .Os }}-
+      {{- if eq .Arch "386" }}i386
+      {{- else }}{{ .Arch }}{{ end }}
+      {{- if .Arm }}v{{ .Arm }}{{ end }}
+    # use zip for windows archives
+    builds: [flowd]
+    format_overrides:
+    - goos: windows
+      format: zip
+    # only embed binaries for now
+    # https://goreleaser.com/customization/archive/?h=archives#packaging-only-the-binaries
+    files:
+    - none*
+
+changelog:
+  use: github-native
+
+# https://goreleaser.com/customization/release/
+release:
+  # Repo in which the release will be created.
+  # Default is extracted from the origin remote URL or empty if its private hosted.
+  github:
+    owner: runreveal
+    name: flow
+
+checksum:
+  name_template: 'checksums.txt'
+
+snapshot:
+  name_template: "{{ incpatch .Version }}-{{ .ShortCommit }}-{{ .Branch }}"
+
+# The lines beneath this are called `modelines`. See `:help modeline`
+# Feel free to remove those if you don't want/use them.
+# yaml-language-server: $schema=https://goreleaser.com/static/schema.json
+# vim: set ts=2 sw=2 tw=0 fo=cnqoj

Makefile

@@ -1,21 +1,34 @@
-VERSION := $(shell git describe --tags --always --dirty="-dev")
+VERSION := $(shell git describe --tags --always --dirty="_dev")
 LDFLAGS := -ldflags='-X "main.version=$(VERSION)"'
-# GCFLAGS := -gcflags='-G=3'
-GO = go
-Q = @
-
-GOTESTFLAGS = -race $(GCFLAGS)
-ifndef Q
-	GOTESTFLAGS += -v
-endif
 
+GOTESTFLAGS = -race
 GOTAGS = testing
 
+GO ?= $(shell which go)
+
+export GOEXPERIMENT=nocoverageredesign
+
 .PHONY: test
-test: vet
-	$Q$(GO) test -vet=off -tags='$(GOTAGS)' $(GOTESTFLAGS) -coverpkg="./..." -coverprofile=.coverprofile ./...
-	$Qgrep -v 'cmd' < .coverprofile > .covprof && mv .covprof .coverprofile
-	$Q$(GO) tool cover -func=.coverprofile
+test: compose
+	$(GO) test -vet=off -tags='$(GOTAGS)' $(GOTESTFLAGS) -coverpkg="./..." -coverprofile=.coverprofile ./...
+	grep -v 'cmd' < .coverprofile > .covprof && mv .covprof .coverprofile
+	$(GO) tool cover -func=.coverprofile
+
+.PHONY: coverage
+coverage:
+	$(GO) tool cover -html=.coverprofile
+
+.PHONY: version
+version:
+	@echo $(VERSION)
+
+.PHONY: dist
+dist:
+	goreleaser release --config .goreleaser.public.yaml --clean --snapshot
+
+.PHONY: compose
+compose:
+	docker-compose up -d
 
 .PHONY: lint
 lint: $(GOPATH)/bin/golangci-lint
@@ -24,19 +37,17 @@ lint: $(GOPATH)/bin/golangci-lint
 $(GOPATH)/bin/golangci-lint:
 	$(GO) install github.com/golangci/golangci-lint/cmd/[email protected]
 
-.PHONY: build
-build: vet
-	$Q$(GO) build $(LDFLAGS) $(GCFLAGS) -o ./build/$(NAME) ./...
-
-.PHONY: vet
-vet:
-	$Q$(GO) vet ./...
+$(GOPATH)/bin/golines:
+	$(GO) install github.com/segmentio/golines@latest
 
 .PHONY: fmtcheck
-fmtchk:
-	$Qexit $(shell goimports -l . | grep -v '^vendor' | wc -l)
+fmtcheck: $(GOPATH)/bin/golines
+	exit $(shell golines -m 128 -l . | wc -l)
 
 .PHONY: fmtfix
-fmtfix:
-	$Qgoimports -w $(shell find . -iname '*.go' | grep -v vendor)
+fmtfix: $(GOPATH)/bin/golines
+	golines -m 128 -w .
 
+.PHONY: clean
+clean:
+	rm -rf dist

README.md

@@ -1,22 +1,34 @@
+# flow
 
+flow is an opinionated framework for scalable, reliable stream processing.
 
-# Support Wishlist & Priorities
+# flowd
 
-Kafka
-redis
+flowd is a daemon for collecting system logs and metrics which makes is powered
+by flow.
 
-socket
-http
+# Installation
 
-amqp
-NATS
+TODO
+
+# Disclaimer
+
+This is nascent software, subject to breaking changes as we reach a good
+working set of APIs, interfaces and data models.
+
+# TODO
+
+- Ensure that consumers of flow aren't subject to all the dependencies of flowd.
+- Consider breaking apart the library from the daemon.
 
+# Source Wishlist
+
+Kafka
+redis
+NATS
+amqp
 pubsub
 Kinesis
-SQS
-
 memcache?
 zmq?
 NSQ?
-NATS?
-

cmd/flowd/config.go

@@ -0,0 +1,84 @@
+package main
+
+import (
+	"os"
+
+	"github.com/runreveal/flow"
+	"github.com/runreveal/flow/internal/destinations"
+	"github.com/runreveal/flow/internal/destinations/runreveal"
+	"github.com/runreveal/flow/internal/sources"
+	"github.com/runreveal/flow/internal/sources/journald"
+	"github.com/runreveal/flow/internal/sources/syslog"
+	"github.com/runreveal/flow/internal/types"
+	"github.com/runreveal/lib/loader"
+	"golang.org/x/exp/slog"
+	// We could register and configure these in a separate package
+	// using the init() function.
+	// That would make it easy to "dynamically" enable and disable them at
+	// compile time since it would simply be updating the import list.
+)
+
+func init() {
+	loader.Register("scanner", func() loader.Builder[flow.Source[types.Event]] {
+		return &ScannerConfig{}
+	})
+	loader.Register("syslog", func() loader.Builder[flow.Source[types.Event]] {
+		return &SyslogConfig{}
+	})
+	loader.Register("journald", func() loader.Builder[flow.Source[types.Event]] {
+		return &JournaldConfig{}
+	})
+
+	loader.Register("printer", func() loader.Builder[flow.Destination[types.Event]] {
+		return &PrinterConfig{}
+	})
+	loader.Register("runreveal", func() loader.Builder[flow.Destination[types.Event]] {
+		return &RunRevealConfig{}
+	})
+}
+
+type ScannerConfig struct {
+}
+
+func (c *ScannerConfig) Configure() (flow.Source[types.Event], error) {
+	slog.Info("configuring scanner")
+	return sources.NewScanner(os.Stdin), nil
+}
+
+type SyslogConfig struct {
+	Addr string `json:"addr"`
+}
+
+func (c *SyslogConfig) Configure() (flow.Source[types.Event], error) {
+	slog.Info("configuring syslog")
+	return syslog.NewSyslogSource(syslog.SyslogCfg{
+		Addr: c.Addr,
+	}), nil
+}
+
+type PrinterConfig struct {
+}
+
+func (c *PrinterConfig) Configure() (flow.Destination[types.Event], error) {
+	slog.Info("configuring printer")
+	return destinations.NewPrinter(os.Stdout), nil
+}
+
+type RunRevealConfig struct {
+	WebhookURL string `json:"webhookURL"`
+}
+
+func (c *RunRevealConfig) Configure() (flow.Destination[types.Event], error) {
+	slog.Info("configuring runreveal")
+	return runreveal.New(
+		runreveal.WithWebhookURL(c.WebhookURL),
+	), nil
+}
+
+type JournaldConfig struct {
+}
+
+func (c *JournaldConfig) Configure() (flow.Source[types.Event], error) {
+	slog.Info("configuring journald")
+	return journald.New(), nil
+}