README.md

@@ -1,7 +1,5 @@
 # simplelog
 
-## Summary
-
 simplelog is an opinionated logging package designed to facilitate easy and
 structured logging in Go applications with an absolute minimum of
 boilerplate.
@@ -9,16 +7,12 @@ boilerplate.
 The idea is that you can add a single import line which replaces the
 stdlib `log/slog` default handler, and solve the 90% case for logging.
 
-## Current Status
-
-Pre-1.0, not working yet.
-
 ## Features
 
--   if output is a tty, outputs pretty color logs
--   if output is not a tty, outputs json
--   supports delivering logs via tcp RELP (e.g. to remote rsyslog using imrelp)
--   supports delivering each log message via a webhook
+* if output is a tty, outputs pretty color logs
+* if output is not a tty, outputs json
+* supports delivering logs via tcp RELP (e.g. to remote rsyslog using imrelp)
+* supports delivering each log message via a webhook
 
 ## Installation
 
@@ -31,7 +25,7 @@ go mod init your_project_name
 Then, add SimpleLog to your project:
 
 ```bash
-go get sneak.berlin/go/simplelog
+go get git.eeqj.de/sneak/go-simplelog
 ```
 
 ## Usage
@@ -43,7 +37,7 @@ package main
 
 import (
 	"log/slog"
-	_ "sneak.berlin/go/simplelog"
+	_ "git.eeqj.de/sneak/go-simplelog"
 )
 
 func main() {

relp_handler.go

@@ -0,0 +1,274 @@
+package simplelog
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"log/slog"
+	"net"
+	"net/url"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+func getEnvAsInt(name string, defaultVal int) int {
+	valStr := os.Getenv(name)
+	if val, err := strconv.Atoi(valStr); err == nil {
+		return val
+	}
+	return defaultVal
+}
+
+func getEnvAsDuration(name string, defaultVal time.Duration) time.Duration {
+	valStr := os.Getenv(name)
+	if val, err := time.ParseDuration(valStr); err == nil {
+		return val
+	}
+	return defaultVal
+}
+
+var (
+	cacheDir, _       = os.UserCacheDir()
+	diskBufferLimit   = getEnvAsInt("LOGGER_DISK_BUFFER_LIMIT", 100)
+	diskWriteInterval = getEnvAsDuration("LOGGER_DISK_WRITE_INTERVAL", time.Second)
+	relpDebug         = os.Getenv("RELP_DEBUG") != ""
+)
+
+type RELPHandler struct {
+	relpServerURL string
+	relpHost      string
+	relpPort      string
+	conn          net.Conn
+	ch            chan Event
+	done          chan struct{}
+	failedCh      chan Event
+	timer         *time.Timer
+}
+
+func NewRELPHandler(relpURL string) (ExtendedHandler, error) {
+	parsedURL, err := url.Parse(relpURL)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing RELP URL: %v", err)
+	}
+	if parsedURL.Scheme != "tcp" {
+		return nil, fmt.Errorf("the RELP URL must have the tcp scheme, got %s", parsedURL.Scheme)
+	}
+
+	host, port, err := net.SplitHostPort(parsedURL.Host)
+	if err != nil {
+		return nil, fmt.Errorf("Error splitting host and port: %v", err)
+	}
+
+	if err := os.MkdirAll(filepath.Join(cacheDir, "simplelog"), 0755); err != nil {
+		return nil, fmt.Errorf("failed to create cache directory: %v", err)
+	}
+	r := &RELPHandler{
+		ch:       make(chan Event, diskBufferLimit),
+		done:     make(chan struct{}),
+		failedCh: make(chan Event, diskBufferLimit),
+		timer:    time.NewTimer(diskWriteInterval),
+		relpHost: host,
+		relpPort: port,
+	}
+	if relpDebug {
+		log.Printf("Created new RELP handler for server at %s", r.relpServerURL)
+	}
+
+	err = r.Startup()
+	if err != nil {
+		return nil, err
+	}
+	return r, nil
+}
+
+func (r *RELPHandler) connectToRELPServer() (net.Conn, error) {
+	conn, err := net.Dial("tcp", net.JoinHostPort(r.relpHost, r.relpPort))
+	if err != nil {
+		if relpDebug {
+			log.Printf("Failed to connect to RELP server at %s: %v", net.JoinHostPort(r.relpHost, r.relpPort), err)
+		}
+		return nil, err
+	}
+	if relpDebug {
+		log.Printf("Successfully connected to RELP server at %s", net.JoinHostPort(r.relpHost, r.relpPort))
+	}
+	return conn, nil
+}
+
+func (r *RELPHandler) Startup() error {
+	var err error
+	r.conn, err = r.connectToRELPServer()
+	if err != nil {
+		if relpDebug {
+			log.Printf("Failed to establish TCP connection to RELP server: %v", err)
+		}
+		return fmt.Errorf("Failed to establish TCP connection to RELP server: %v", err)
+	}
+	if relpDebug {
+		log.Printf("Successfully connected to RELP server at %s", r.relpServerURL)
+	}
+	go r.receiveEventsFromChannel()
+	go r.processFailedEvents()
+	go r.watchDirectoryForFailedEventFiles()
+	return nil
+}
+
+func (r *RELPHandler) Enabled(ctx context.Context, level slog.Level) bool {
+	return true
+}
+
+func (r *RELPHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	return r
+}
+
+func (r *RELPHandler) WithGroup(name string) slog.Handler {
+	return r
+}
+
+func (r *RELPHandler) Handle(ctx context.Context, record slog.Record) error {
+	attrs := make(map[string]interface{})
+	record.Attrs(func(attr slog.Attr) bool {
+		attrs[attr.Key] = attr.Value
+		return true
+	})
+	jsonData, err := json.Marshal(attrs)
+	if err != nil {
+		return fmt.Errorf("error marshaling attributes: %v", err)
+	}
+	// Get the caller information
+	_, file, line, ok := runtime.Caller(5)
+	if !ok {
+		file = "???"
+		line = 0
+	}
+
+	event := NewExtendedEvent(record.Level.String(), record.Message, jsonData, file, line)
+	for _, handler := range cl.handlers {
+		if err := handler.Handle(ctx, event); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *RELPHandler) receiveEventsFromChannel() {
+	for {
+		select {
+		case event := <-r.ch:
+			if err := r.sendEventToRELPServer(event); err != nil {
+				r.failedCh <- event
+			}
+		case <-r.done:
+			return
+		}
+	}
+}
+
+func (r *RELPHandler) processFailedEvents() {
+	for {
+		select {
+		case event := <-r.failedCh:
+			if err := r.sendEventToRELPServer(event); err != nil {
+				// If still failing, write to disk
+				r.writeFailedToDisk(event)
+			}
+		case <-r.timer.C:
+			// Flush all events in failedCh to disk
+			for len(r.failedCh) > 0 {
+				event := <-r.failedCh
+				r.writeFailedToDisk(event)
+			}
+		case <-r.done:
+			return
+		}
+	}
+}
+
+func (r *RELPHandler) sendEventToRELPServer(event Event) error {
+	jsonData, err := json.Marshal(event)
+	if err != nil {
+		return fmt.Errorf("error marshaling event: %v", err)
+	}
+	_, err = r.conn.Write(jsonData)
+	if err != nil {
+		return err
+	}
+	// Implement TCP read for the acknowledgment with a timeout
+	r.conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	ack := make([]byte, 256)
+	n, err := r.conn.Read(ack)
+	if err != nil {
+		return err
+	}
+	if string(ack[:n]) != "ACK" {
+		return fmt.Errorf("expected ACK from server, got %s", string(ack[:n]))
+	}
+	if relpDebug {
+		log.Printf("Received ACK from RELP server for event %s", event.ID)
+	}
+	return nil
+}
+
+func (r *RELPHandler) watchDirectoryForFailedEventFiles() {
+	for {
+		select {
+		case <-r.done:
+			return
+		default:
+			if r.isConnected() {
+				if relpDebug {
+					log.Printf("Reconnected to RELP server at %s", r.relpServerURL)
+				}
+				files, err := ioutil.ReadDir(cacheDir)
+				if err != nil {
+					log.Printf("Error reading cache directory: %v", err)
+					continue
+				}
+				for _, file := range files {
+					filePath := filepath.Join(cacheDir, file.Name())
+					data, err := ioutil.ReadFile(filePath)
+					if err != nil {
+						log.Printf("Error reading event cache file %s: %v", file.Name(), err)
+						continue
+					}
+					var event Event
+					if err := json.Unmarshal(data, &event); err != nil {
+						log.Printf("Error unmarshalling event from file %s: %v", file.Name(), err)
+						continue
+					}
+					r.ch <- event
+					os.Remove(filePath) // Remove file after processing
+				}
+			}
+			time.Sleep(10 * time.Second) // Check disk every 10 seconds
+		}
+	}
+}
+
+func (r *RELPHandler) isConnected() bool {
+	_, err := r.conn.Write([]byte{})
+	return err == nil
+}
+
+func (r *RELPHandler) Shutdown() error {
+	close(r.done)
+	if r.conn != nil {
+		if err := r.conn.Close(); err != nil {
+			return fmt.Errorf("error closing TCP connection: %v", err)
+		}
+	}
+	return nil
+}
+
+func (r *RELPHandler) writeFailedToDisk(event Event) {
+	fileName := filepath.Join(cacheDir, "simplelog", uuid.New().String()+".logevent")
+	data, _ := json.Marshal(event)
+	ioutil.WriteFile(fileName, data, 0600)
+}

simplelog.go

@@ -2,18 +2,16 @@ package simplelog
 
 import (
 	"context"
-	"encoding/json"
 	"log"
+	"runtime"
 	"log/slog"
 	"os"
-	"time"
-
-	"github.com/google/uuid"
 
 	"github.com/mattn/go-isatty"
 )
 
 var (
+	relpServerURL = os.Getenv("LOGGER_RELP_URL")
 	webhookURL    = os.Getenv("LOGGER_WEBHOOK_URL")
 )
 
@@ -37,6 +35,13 @@ func NewMultiplexHandler() slog.Handler {
 	} else {
 		cl.handlers = append(cl.handlers, NewJSONHandler())
 	}
+	if relpServerURL != "" {
+		handler, err := NewRELPHandler(relpServerURL)
+		if err != nil {
+			log.Fatalf("Failed to initialize RELP handler: %v", err)
+		}
+		cl.handlers = append(cl.handlers, handler)
+	}
 	if webhookURL != "" {
 		handler, err := NewWebhookHandler(webhookURL)
 		if err != nil {
@@ -47,10 +52,7 @@ func NewMultiplexHandler() slog.Handler {
 	return cl
 }
 
-func (cl *MultiplexHandler) Handle(
-	ctx context.Context,
-	record slog.Record,
-) error {
+func (cl *MultiplexHandler) Handle(ctx context.Context, record slog.Record) error {
 	for _, handler := range cl.handlers {
 		if err := handler.Handle(ctx, record); err != nil {
 			return err
@@ -59,10 +61,7 @@ func (cl *MultiplexHandler) Handle(
 	return nil
 }
 
-func (cl *MultiplexHandler) Enabled(
-	ctx context.Context,
-	level slog.Level,
-) bool {
+func (cl *MultiplexHandler) Enabled(ctx context.Context, level slog.Level) bool {
 	// send us all events
 	return true
 }
@@ -82,7 +81,6 @@ func (cl *MultiplexHandler) WithGroup(name string) slog.Handler {
 	}
 	return &MultiplexHandler{handlers: newHandlers}
 }
-
 type ExtendedEvent interface {
 	GetID() uuid.UUID
 	GetTimestamp() time.Time

tools/relp_log_trial.go

@@ -3,7 +3,7 @@ package main
 import (
 	"log/slog"
 
-	_ "sneak.berlin/go/simplelog" // Using underscore to only invoke init()
+	_ "git.eeqj.de/sneak/go-simplelog" // Using underscore to only invoke init()
 )
 
 func main() {