revamped fileSource+amqpSink (#270)

controller/metrics2/amqpSink.go

@@ -0,0 +1,64 @@
+package metrics2
+
+import (
+	"context"
+	"github.com/michaelquigley/cf"
+	"github.com/openziti/zrok/controller/env"
+	"github.com/pkg/errors"
+	amqp "github.com/rabbitmq/amqp091-go"
+	"time"
+)
+
+func init() {
+	env.GetCfOptions().AddFlexibleSetter("amqpSink", loadAmqpSinkConfig)
+}
+
+type AmqpSinkConfig struct {
+	Url       string `cf:"+secret"`
+	QueueName string
+}
+
+func loadAmqpSinkConfig(v interface{}, _ *cf.Options) (interface{}, error) {
+	if submap, ok := v.(map[string]interface{}); ok {
+		cfg := &AmqpSinkConfig{}
+		if err := cf.Bind(cfg, submap, cf.DefaultOptions()); err != nil {
+			return nil, err
+		}
+		return newAmqpSink(cfg)
+	}
+	return nil, errors.New("invalid config structure for 'amqpSink'")
+}
+
+type amqpSink struct {
+	conn  *amqp.Connection
+	ch    *amqp.Channel
+	queue amqp.Queue
+}
+
+func newAmqpSink(cfg *AmqpSinkConfig) (*amqpSink, error) {
+	conn, err := amqp.Dial(cfg.Url)
+	if err != nil {
+		return nil, errors.Wrap(err, "error dialing amqp broker")
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting amqp channel")
+	}
+
+	queue, err := ch.QueueDeclare(cfg.QueueName, true, false, false, false, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "error declaring queue")
+	}
+
+	return &amqpSink{conn, ch, queue}, nil
+}
+
+func (s *amqpSink) Handle(event ZitiEventJson) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	return s.ch.PublishWithContext(ctx, "", s.queue.Name, false, false, amqp.Publishing{
+		ContentType: "application/json",
+		Body:        []byte(event),
+	})
+}

controller/metrics2/fileSource.go

@@ -0,0 +1,130 @@
+package metrics2
+
+import (
+	"encoding/binary"
+	"github.com/michaelquigley/cf"
+	"github.com/nxadm/tail"
+	"github.com/openziti/zrok/controller/env"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"os"
+)
+
+func init() {
+	env.GetCfOptions().AddFlexibleSetter("fileSource", loadFileSourceConfig)
+}
+
+type FileSourceConfig struct {
+	Path        string
+	PointerPath string
+}
+
+func loadFileSourceConfig(v interface{}, _ *cf.Options) (interface{}, error) {
+	if submap, ok := v.(map[string]interface{}); ok {
+		cfg := &FileSourceConfig{}
+		if err := cf.Bind(cfg, submap, cf.DefaultOptions()); err != nil {
+			return nil, err
+		}
+		return &fileSource{cfg: cfg}, nil
+	}
+	return nil, errors.New("invalid config structure for 'fileSource'")
+}
+
+type fileSource struct {
+	cfg  *FileSourceConfig
+	ptrF *os.File
+	t    *tail.Tail
+}
+
+func (s *fileSource) Start(events chan ZitiEventJson) (join chan struct{}, err error) {
+	f, err := os.Open(s.cfg.Path)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error opening '%v'", s.cfg.Path)
+	}
+	_ = f.Close()
+
+	s.ptrF, err = os.OpenFile(s.pointerPath(), os.O_CREATE|os.O_RDWR, os.ModePerm)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error opening pointer '%v'", s.pointerPath())
+	}
+
+	ptr, err := s.readPtr()
+	if err != nil {
+		return nil, errors.Wrap(err, "error reading pointer")
+	}
+	logrus.Infof("retrieved stored position pointer at '%d'", ptr)
+
+	join = make(chan struct{})
+	go func() {
+		s.tail(ptr, events)
+		close(join)
+	}()
+
+	return join, nil
+}
+
+func (s *fileSource) Stop() {
+	if err := s.t.Stop(); err != nil {
+		logrus.Error(err)
+	}
+}
+
+func (s *fileSource) tail(ptr int64, events chan ZitiEventJson) {
+	logrus.Info("started")
+	defer logrus.Info("stopped")
+
+	var err error
+	s.t, err = tail.TailFile(s.cfg.Path, tail.Config{
+		ReOpen:   true,
+		Follow:   true,
+		Location: &tail.SeekInfo{Offset: ptr},
+	})
+	if err != nil {
+		logrus.Error(err)
+		return
+	}
+
+	for event := range s.t.Lines {
+		events <- ZitiEventJson(event.Text)
+
+		if err := s.writePtr(event.SeekInfo.Offset); err != nil {
+			logrus.Error(err)
+		}
+	}
+}
+
+func (s *fileSource) pointerPath() string {
+	if s.cfg.PointerPath == "" {
+		return s.cfg.Path + ".ptr"
+	} else {
+		return s.cfg.PointerPath
+	}
+}
+
+func (s *fileSource) readPtr() (int64, error) {
+	ptr := int64(0)
+	buf := make([]byte, 8)
+	if n, err := s.ptrF.Seek(0, 0); err == nil && n == 0 {
+		if n, err := s.ptrF.Read(buf); err == nil && n == 8 {
+			ptr = int64(binary.LittleEndian.Uint64(buf))
+			return ptr, nil
+		} else {
+			return -1, errors.Wrapf(err, "error reading pointer (%d): %v", n, err)
+		}
+	} else {
+		return -1, errors.Wrapf(err, "error seeking pointer (%d): %v", n, err)
+	}
+}
+
+func (s *fileSource) writePtr(ptr int64) error {
+	buf := make([]byte, 8)
+	binary.LittleEndian.PutUint64(buf, uint64(ptr))
+	if n, err := s.ptrF.Seek(0, 0); err == nil && n == 0 {
+		if n, err := s.ptrF.Write(buf); err != nil || n != 8 {
+			return errors.Wrapf(err, "error writing pointer (%d): %v", n, err)
+		}
+	} else {
+		return errors.Wrapf(err, "error seeking pointer (%d): %v", n, err)
+	}
+	return nil
+}

controller/metrics2/model.go

@@ -0,0 +1,12 @@
+package metrics2
+
+type ZitiEventJson string
+
+type ZitiEventJsonSource interface {
+	Start(chan ZitiEventJson) (join chan struct{}, err error)
+	Stop()
+}
+
+type ZitiEventJsonSink interface {
+	Handle(event ZitiEventJson) error
+}