Add support for running systemd units

go.mod

@@ -4,20 +4,19 @@ go 1.20
 
 require (
 	git.sr.ht/~blallo/logz/interface v0.0.0-20220324191132-95d94ae8e337
+	git.sr.ht/~blallo/logz/testlogger v0.0.0-20230212191205-53d5ce2c0d54
+	git.sr.ht/~blallo/logz/zlog v0.0.0-20220324191132-95d94ae8e337
 	github.com/cenkalti/backoff/v4 v4.2.0
+	github.com/coreos/go-systemd/v22 v22.5.0
+	github.com/go-chi/chi/v5 v5.0.8
+	github.com/gorilla/websocket v1.5.0
+	github.com/stretchr/testify v1.8.1
 )
 
 require (
-	git.sr.ht/~blallo/logz v0.0.0-20220324191132-95d94ae8e337 // indirect
-	git.sr.ht/~blallo/logz/testlogger v0.0.0-20230212191205-53d5ce2c0d54 // indirect
-	git.sr.ht/~blallo/logz/zlog v0.0.0-20220324191132-95d94ae8e337 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/go-chi/chi/v5 v5.0.8 // indirect
+	github.com/godbus/dbus/v5 v5.0.4 // indirect
-	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rs/zerolog v1.26.1 // indirect
-	github.com/stretchr/objx v0.5.0 // indirect
-	github.com/stretchr/testify v1.8.1 // indirect
-	golang.org/x/net v0.6.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,5 +1,3 @@
-git.sr.ht/~blallo/logz v0.0.0-20220324191132-95d94ae8e337 h1:QYgqHKnaExscRlUrQvpm9AiHnpOVJnoQBw/W+VCZaK8=
-git.sr.ht/~blallo/logz v0.0.0-20220324191132-95d94ae8e337/go.mod h1:W/OSkm9pxF84geA9tK+A9Ys028UF1ayD41BJlCDC4Io=
 git.sr.ht/~blallo/logz/interface v0.0.0-20220324191132-95d94ae8e337 h1:a62rvbRTBnosIciC9Bg7i8XlpYDmMDGecj1WCUgg8Uo=
 git.sr.ht/~blallo/logz/interface v0.0.0-20220324191132-95d94ae8e337/go.mod h1:V1e+pLie6GMc2iEdyhB3+bSfFBvwY0qDcQmyIQ3Jr3I=
 git.sr.ht/~blallo/logz/testlogger v0.0.0-20230212191205-53d5ce2c0d54 h1:CHxlq5zroyZORIG9mFvAxLTNuMyUFZZGStevHvAF+ek=
@@ -9,11 +7,14 @@ git.sr.ht/~blallo/logz/zlog v0.0.0-20220324191132-95d94ae8e337/go.mod h1:9fUP8io
 github.com/cenkalti/backoff/v4 v4.2.0 h1:HN5dHm3WBOgndBH6E8V0q2jIYIR3s9yglV8k/+MN3u4=
 github.com/cenkalti/backoff/v4 v4.2.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
+github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-chi/chi/v5 v5.0.8 h1:lD+NLqFcAi1ovnVZpsnObHGW4xb4J8lNmoYVfECH1Y0=
 github.com/go-chi/chi/v5 v5.0.8/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
+github.com/godbus/dbus/v5 v5.0.4 h1:9349emZab16e7zQvpmsbtjc18ykshndd8y2PG3sgJbA=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
@@ -25,7 +26,6 @@ github.com/rs/zerolog v1.26.1 h1:/ihwxqH+4z8UxyI70wM1z9yCvkWcfz/a3mj48k/Zngc=
 github.com/rs/zerolog v1.26.1/go.mod h1:/wSSJWX7lVrsOwlbyTRSOJvqRlc+WjWlfes+CiJ+tmc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
@@ -40,8 +40,6 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.6.0 h1:L4ZwwTvKW9gr0ZMS1yrHD9GZhIuVjOBBnaKH+SPQK0Q=
-golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -59,6 +57,7 @@ golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

systemd.go

@@ -0,0 +1,175 @@
+package broadcast
+
+import (
+	"context"
+	"io/ioutil"
+	"strings"
+
+	"github.com/coreos/go-systemd/v22/dbus"
+	"github.com/coreos/go-systemd/v22/sdjournal"
+
+	"git.sr.ht/~blallo/logz/interface"
+)
+
+var _ Runnable = &SystemdUnit{}
+
+type SystemdUnit struct {
+	unit   string
+	user   bool
+	conn   *dbus.Conn
+	logger logz.Logger
+}
+
+func NewSystemdUnit(logger logz.Logger, unit string, user bool) *SystemdUnit {
+	return &SystemdUnit{
+		unit:   unit,
+		user:   user,
+		logger: logger,
+	}
+}
+
+func (u *SystemdUnit) Init(ctx context.Context) error {
+	connCh := make(chan *dbus.Conn)
+	errCh := make(chan error)
+
+	go func() {
+		var conn *dbus.Conn
+		var err error
+
+		if u.user {
+			conn, err = dbus.NewUserConnection()
+		} else {
+			conn, err = dbus.NewSystemConnection()
+		}
+		if err != nil {
+			errCh <- err
+		}
+		connCh <- conn
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-errCh:
+		return err
+	case conn := <-connCh:
+		u.conn = conn
+		return nil
+	}
+}
+
+func (u *SystemdUnit) Start(ctx context.Context) error {
+	resultCh := make(chan string)
+	errCh := make(chan error)
+
+	go func() {
+		_, err := u.conn.StartUnit(u.unit, "fail", resultCh)
+		if err != nil {
+			errCh <- err
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-errCh:
+		return err
+	case result := <-resultCh:
+		switch result {
+		case "done":
+			return nil
+		default:
+			u.logger.Warn(map[string]any{
+				"msg":    "The unit failed to start",
+				"reason": result,
+			})
+			return ErrNotRunning
+		}
+	}
+}
+
+func (u *SystemdUnit) Stop(ctx context.Context) error {
+	resultCh := make(chan string)
+	errCh := make(chan error)
+
+	go func() {
+		_, err := u.conn.StopUnit(u.unit, "fail", resultCh)
+		if err != nil {
+			errCh <- err
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-errCh:
+		return err
+	case result := <-resultCh:
+		switch result {
+		case "done":
+			return nil
+		default:
+			u.logger.Warn(map[string]any{
+				"msg":    "The unit failed to stop",
+				"reason": result,
+			})
+			return ErrCannotStop
+		}
+	}
+}
+
+func (u *SystemdUnit) Logs(context.Context) []string {
+	journal, err := sdjournal.NewJournalReader(sdjournal.JournalReaderConfig{
+		NumFromTail: bufLines,
+		Matches: []sdjournal.Match{
+			{
+				Field: sdjournal.SD_JOURNAL_FIELD_SYSTEMD_UNIT,
+				Value: u.unit,
+			},
+		},
+	})
+	if err != nil {
+		u.logger.Err(map[string]any{
+			"msg": "Cannot create journal reader",
+			"err": err.Error(),
+		})
+		return nil
+	}
+
+	res, err := ioutil.ReadAll(journal)
+	if err != nil {
+		u.logger.Err(map[string]any{
+			"msg": "Cannot read journal",
+			"err": err.Error(),
+		})
+		return nil
+	}
+
+	// XXX: this is suboptimal, as it goes over the whole res twice.
+	return strings.Split(string(res), "\n")
+}
+
+func (u *SystemdUnit) Liveness(ctx context.Context) error {
+	status, err := u.conn.ListUnitsByNames([]string{u.unit})
+	if err != nil {
+		return err
+	}
+
+	if len(status) != 1 {
+		u.logger.Warn(map[string]any{
+			"msg":    "Unexpected status",
+			"status": status,
+		})
+		return ErrNotRunning
+	}
+
+	if status[0].ActiveState == "active" {
+		return nil
+	}
+
+	u.logger.Debug(map[string]any{
+		"msg":    "Status is not active",
+		"status": status[0].ActiveState,
+	})
+	return ErrNotRunning
+}