broadcast/systemd.go

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
}