From 68ce353a23c8cd15077c70bc730470841ef37fb7 Mon Sep 17 00:00:00 2001 From: Peter Hunt Date: Thu, 23 May 2019 14:52:50 -0400 Subject: bump go-systemd version Signed-off-by: Peter Hunt --- vendor/github.com/coreos/go-systemd/NOTICE | 5 + vendor/github.com/coreos/go-systemd/README.md | 21 ++- .../coreos/go-systemd/activation/files.go | 19 ++- .../coreos/go-systemd/activation/listeners.go | 51 ++++++- .../coreos/go-systemd/activation/packetconns.go | 5 +- vendor/github.com/coreos/go-systemd/dbus/dbus.go | 31 +++- .../github.com/coreos/go-systemd/dbus/methods.go | 43 ++++-- vendor/github.com/coreos/go-systemd/dbus/set.go | 2 +- .../coreos/go-systemd/dbus/subscription.go | 157 ++++++++++++++++----- .../coreos/go-systemd/journal/journal.go | 7 +- .../coreos/go-systemd/sdjournal/journal.go | 134 +++++++++++++++--- .../github.com/coreos/go-systemd/sdjournal/read.go | 124 ++++++++-------- 12 files changed, 463 insertions(+), 136 deletions(-) create mode 100644 vendor/github.com/coreos/go-systemd/NOTICE (limited to 'vendor') diff --git a/vendor/github.com/coreos/go-systemd/NOTICE b/vendor/github.com/coreos/go-systemd/NOTICE new file mode 100644 index 000000000..23a0ada2f --- /dev/null +++ b/vendor/github.com/coreos/go-systemd/NOTICE @@ -0,0 +1,5 @@ +CoreOS Project +Copyright 2018 CoreOS, Inc + +This product includes software developed at CoreOS, Inc. +(http://www.coreos.com/). diff --git a/vendor/github.com/coreos/go-systemd/README.md b/vendor/github.com/coreos/go-systemd/README.md index cb87a1124..cad04a803 100644 --- a/vendor/github.com/coreos/go-systemd/README.md +++ b/vendor/github.com/coreos/go-systemd/README.md @@ -6,9 +6,11 @@ Go bindings to systemd. The project has several packages: - `activation` - for writing and using socket activation from Go +- `daemon` - for notifying systemd of service status changes - `dbus` - for starting/stopping/inspecting running services and units - `journal` - for writing to systemd's logging service, journald - `sdjournal` - for reading from journald by wrapping its C API +- `login1` - for integration with the systemd logind API - `machine1` - for registering machines/containers with systemd - `unit` - for (de)serialization and comparison of unit files @@ -18,10 +20,9 @@ An example HTTP server using socket activation can be quickly set up by followin https://github.com/coreos/go-systemd/tree/master/examples/activation/httpserver -## Journal +## systemd Service Notification -Using the pure-Go `journal` package you can submit journal entries directly to systemd's journal, taking advantage of features like indexed key/value pairs for each log entry. -The `sdjournal` package provides read access to the journal by wrapping around journald's native C API; consequently it requires cgo and the journal headers to be available. +The `daemon` package is an implementation of the [sd_notify protocol](https://www.freedesktop.org/software/systemd/man/sd_notify.html#Description). It can be used to inform systemd of service start-up completion, watchdog events, and other status changes. ## D-Bus @@ -45,6 +46,20 @@ Create `/etc/dbus-1/system-local.conf` that looks like this: ``` +## Journal + +### Writing to the Journal + +Using the pure-Go `journal` package you can submit journal entries directly to systemd's journal, taking advantage of features like indexed key/value pairs for each log entry. + +### Reading from the Journal + +The `sdjournal` package provides read access to the journal by wrapping around journald's native C API; consequently it requires cgo and the journal headers to be available. + +## logind + +The `login1` package provides functions to integrate with the [systemd logind API](http://www.freedesktop.org/wiki/Software/systemd/logind/). + ## machined The `machine1` package allows interaction with the [systemd machined D-Bus API](http://www.freedesktop.org/wiki/Software/systemd/machined/). diff --git a/vendor/github.com/coreos/go-systemd/activation/files.go b/vendor/github.com/coreos/go-systemd/activation/files.go index c8e85fcd5..29dd18def 100644 --- a/vendor/github.com/coreos/go-systemd/activation/files.go +++ b/vendor/github.com/coreos/go-systemd/activation/files.go @@ -18,18 +18,26 @@ package activation import ( "os" "strconv" + "strings" "syscall" ) -// based on: https://gist.github.com/alberts/4640792 const ( + // listenFdsStart corresponds to `SD_LISTEN_FDS_START`. listenFdsStart = 3 ) +// Files returns a slice containing a `os.File` object for each +// file descriptor passed to this process via systemd fd-passing protocol. +// +// The order of the file descriptors is preserved in the returned slice. +// `unsetEnv` is typically set to `true` in order to avoid clashes in +// fd usage and to avoid leaking environment flags to child processes. func Files(unsetEnv bool) []*os.File { if unsetEnv { defer os.Unsetenv("LISTEN_PID") defer os.Unsetenv("LISTEN_FDS") + defer os.Unsetenv("LISTEN_FDNAMES") } pid, err := strconv.Atoi(os.Getenv("LISTEN_PID")) @@ -42,10 +50,17 @@ func Files(unsetEnv bool) []*os.File { return nil } + names := strings.Split(os.Getenv("LISTEN_FDNAMES"), ":") + files := make([]*os.File, 0, nfds) for fd := listenFdsStart; fd < listenFdsStart+nfds; fd++ { syscall.CloseOnExec(fd) - files = append(files, os.NewFile(uintptr(fd), "LISTEN_FD_"+strconv.Itoa(fd))) + name := "LISTEN_FD_" + strconv.Itoa(fd) + offset := fd - listenFdsStart + if offset < len(names) && len(names[offset]) > 0 { + name = names[offset] + } + files = append(files, os.NewFile(uintptr(fd), name)) } return files diff --git a/vendor/github.com/coreos/go-systemd/activation/listeners.go b/vendor/github.com/coreos/go-systemd/activation/listeners.go index fd5dfc709..bb5cc2311 100644 --- a/vendor/github.com/coreos/go-systemd/activation/listeners.go +++ b/vendor/github.com/coreos/go-systemd/activation/listeners.go @@ -25,13 +25,33 @@ import ( // The order of the file descriptors is preserved in the returned slice. // Nil values are used to fill any gaps. For example if systemd were to return file descriptors // corresponding with "udp, tcp, tcp", then the slice would contain {nil, net.Listener, net.Listener} -func Listeners(unsetEnv bool) ([]net.Listener, error) { - files := Files(unsetEnv) +func Listeners() ([]net.Listener, error) { + files := Files(true) listeners := make([]net.Listener, len(files)) for i, f := range files { if pc, err := net.FileListener(f); err == nil { listeners[i] = pc + f.Close() + } + } + return listeners, nil +} + +// ListenersWithNames maps a listener name to a set of net.Listener instances. +func ListenersWithNames() (map[string][]net.Listener, error) { + files := Files(true) + listeners := map[string][]net.Listener{} + + for _, f := range files { + if pc, err := net.FileListener(f); err == nil { + current, ok := listeners[f.Name()] + if !ok { + listeners[f.Name()] = []net.Listener{pc} + } else { + listeners[f.Name()] = append(current, pc) + } + f.Close() } } return listeners, nil @@ -40,8 +60,8 @@ func Listeners(unsetEnv bool) ([]net.Listener, error) { // TLSListeners returns a slice containing a net.listener for each matching TCP socket type // passed to this process. // It uses default Listeners func and forces TCP sockets handlers to use TLS based on tlsConfig. -func TLSListeners(unsetEnv bool, tlsConfig *tls.Config) ([]net.Listener, error) { - listeners, err := Listeners(unsetEnv) +func TLSListeners(tlsConfig *tls.Config) ([]net.Listener, error) { + listeners, err := Listeners() if listeners == nil || err != nil { return nil, err @@ -58,3 +78,26 @@ func TLSListeners(unsetEnv bool, tlsConfig *tls.Config) ([]net.Listener, error) return listeners, err } + +// TLSListenersWithNames maps a listener name to a net.Listener with +// the associated TLS configuration. +func TLSListenersWithNames(tlsConfig *tls.Config) (map[string][]net.Listener, error) { + listeners, err := ListenersWithNames() + + if listeners == nil || err != nil { + return nil, err + } + + if tlsConfig != nil && err == nil { + for _, ll := range listeners { + // Activate TLS only for TCP sockets + for i, l := range ll { + if l.Addr().Network() == "tcp" { + ll[i] = tls.NewListener(l, tlsConfig) + } + } + } + } + + return listeners, err +} diff --git a/vendor/github.com/coreos/go-systemd/activation/packetconns.go b/vendor/github.com/coreos/go-systemd/activation/packetconns.go index 48b2ca029..a97206785 100644 --- a/vendor/github.com/coreos/go-systemd/activation/packetconns.go +++ b/vendor/github.com/coreos/go-systemd/activation/packetconns.go @@ -24,13 +24,14 @@ import ( // The order of the file descriptors is preserved in the returned slice. // Nil values are used to fill any gaps. For example if systemd were to return file descriptors // corresponding with "udp, tcp, udp", then the slice would contain {net.PacketConn, nil, net.PacketConn} -func PacketConns(unsetEnv bool) ([]net.PacketConn, error) { - files := Files(unsetEnv) +func PacketConns() ([]net.PacketConn, error) { + files := Files(true) conns := make([]net.PacketConn, len(files)) for i, f := range files { if pc, err := net.FilePacketConn(f); err == nil { conns[i] = pc + f.Close() } } return conns, nil diff --git a/vendor/github.com/coreos/go-systemd/dbus/dbus.go b/vendor/github.com/coreos/go-systemd/dbus/dbus.go index c1694fb52..1d54810af 100644 --- a/vendor/github.com/coreos/go-systemd/dbus/dbus.go +++ b/vendor/github.com/coreos/go-systemd/dbus/dbus.go @@ -16,6 +16,7 @@ package dbus import ( + "encoding/hex" "fmt" "os" "strconv" @@ -60,6 +61,27 @@ func PathBusEscape(path string) string { return string(n) } +// pathBusUnescape is the inverse of PathBusEscape. +func pathBusUnescape(path string) string { + if path == "_" { + return "" + } + n := []byte{} + for i := 0; i < len(path); i++ { + c := path[i] + if c == '_' && i+2 < len(path) { + res, err := hex.DecodeString(path[i+1 : i+3]) + if err == nil { + n = append(n, res...) + } + i += 2 + } else { + n = append(n, c) + } + } + return string(n) +} + // Conn is a connection to systemd's dbus endpoint. type Conn struct { // sysconn/sysobj are only used to call dbus methods @@ -74,13 +96,18 @@ type Conn struct { jobs map[dbus.ObjectPath]chan<- string sync.Mutex } - subscriber struct { + subStateSubscriber struct { updateCh chan<- *SubStateUpdate errCh chan<- error sync.Mutex ignore map[dbus.ObjectPath]int64 cleanIgnore int64 } + propertiesSubscriber struct { + updateCh chan<- *PropertiesUpdate + errCh chan<- error + sync.Mutex + } } // New establishes a connection to any available bus and authenticates. @@ -152,7 +179,7 @@ func NewConnection(dialBus func() (*dbus.Conn, error)) (*Conn, error) { sigobj: systemdObject(sigconn), } - c.subscriber.ignore = make(map[dbus.ObjectPath]int64) + c.subStateSubscriber.ignore = make(map[dbus.ObjectPath]int64) c.jobListener.jobs = make(map[dbus.ObjectPath]chan<- string) // Setup the listeners on jobs so that we can get completions diff --git a/vendor/github.com/coreos/go-systemd/dbus/methods.go b/vendor/github.com/coreos/go-systemd/dbus/methods.go index ab17f7cc7..0b4207229 100644 --- a/vendor/github.com/coreos/go-systemd/dbus/methods.go +++ b/vendor/github.com/coreos/go-systemd/dbus/methods.go @@ -1,4 +1,4 @@ -// Copyright 2015 CoreOS, Inc. +// Copyright 2015, 2018 CoreOS, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -16,6 +16,7 @@ package dbus import ( "errors" + "fmt" "path" "strconv" @@ -148,14 +149,27 @@ func (c *Conn) ResetFailedUnit(name string) error { return c.sysobj.Call("org.freedesktop.systemd1.Manager.ResetFailedUnit", 0, name).Store() } -// getProperties takes the unit name and returns all of its dbus object properties, for the given dbus interface -func (c *Conn) getProperties(unit string, dbusInterface string) (map[string]interface{}, error) { +// SystemState returns the systemd state. Equivalent to `systemctl is-system-running`. +func (c *Conn) SystemState() (*Property, error) { + var err error + var prop dbus.Variant + + obj := c.sysconn.Object("org.freedesktop.systemd1", "/org/freedesktop/systemd1") + err = obj.Call("org.freedesktop.DBus.Properties.Get", 0, "org.freedesktop.systemd1.Manager", "SystemState").Store(&prop) + if err != nil { + return nil, err + } + + return &Property{Name: "SystemState", Value: prop}, nil +} + +// getProperties takes the unit path and returns all of its dbus object properties, for the given dbus interface +func (c *Conn) getProperties(path dbus.ObjectPath, dbusInterface string) (map[string]interface{}, error) { var err error var props map[string]dbus.Variant - path := unitPath(unit) if !path.IsValid() { - return nil, errors.New("invalid unit name: " + unit) + return nil, fmt.Errorf("invalid unit name: %v", path) } obj := c.sysconn.Object("org.freedesktop.systemd1", path) @@ -172,9 +186,15 @@ func (c *Conn) getProperties(unit string, dbusInterface string) (map[string]inte return out, nil } -// GetUnitProperties takes the unit name and returns all of its dbus object properties. +// GetUnitProperties takes the (unescaped) unit name and returns all of its dbus object properties. func (c *Conn) GetUnitProperties(unit string) (map[string]interface{}, error) { - return c.getProperties(unit, "org.freedesktop.systemd1.Unit") + path := unitPath(unit) + return c.getProperties(path, "org.freedesktop.systemd1.Unit") +} + +// GetUnitProperties takes the (escaped) unit path and returns all of its dbus object properties. +func (c *Conn) GetUnitPathProperties(path dbus.ObjectPath) (map[string]interface{}, error) { + return c.getProperties(path, "org.freedesktop.systemd1.Unit") } func (c *Conn) getProperty(unit string, dbusInterface string, propertyName string) (*Property, error) { @@ -208,7 +228,8 @@ func (c *Conn) GetServiceProperty(service string, propertyName string) (*Propert // Valid values for unitType: Service, Socket, Target, Device, Mount, Automount, Snapshot, Timer, Swap, Path, Slice, Scope // return "dbus.Error: Unknown interface" if the unitType is not the correct type of the unit func (c *Conn) GetUnitTypeProperties(unit string, unitType string) (map[string]interface{}, error) { - return c.getProperties(unit, "org.freedesktop.systemd1."+unitType) + path := unitPath(unit) + return c.getProperties(path, "org.freedesktop.systemd1."+unitType) } // SetUnitProperties() may be used to modify certain unit properties at runtime. @@ -292,6 +313,7 @@ func (c *Conn) ListUnitsByPatterns(states []string, patterns []string) ([]UnitSt // names and returns an UnitStatus array. Comparing to ListUnitsByPatterns // method, this method returns statuses even for inactive or non-existing // units. Input array should contain exact unit names, but not patterns. +// Note: Requires systemd v230 or higher func (c *Conn) ListUnitsByNames(units []string) ([]UnitStatus, error) { return c.listUnitsInternal(c.sysobj.Call("org.freedesktop.systemd1.Manager.ListUnitsByNames", 0, units).Store) } @@ -563,3 +585,8 @@ func (c *Conn) Reload() error { func unitPath(name string) dbus.ObjectPath { return dbus.ObjectPath("/org/freedesktop/systemd1/unit/" + PathBusEscape(name)) } + +// unitName returns the unescaped base element of the supplied escaped path +func unitName(dpath dbus.ObjectPath) string { + return pathBusUnescape(path.Base(string(dpath))) +} diff --git a/vendor/github.com/coreos/go-systemd/dbus/set.go b/vendor/github.com/coreos/go-systemd/dbus/set.go index f92e6fbed..17c5d4856 100644 --- a/vendor/github.com/coreos/go-systemd/dbus/set.go +++ b/vendor/github.com/coreos/go-systemd/dbus/set.go @@ -36,7 +36,7 @@ func (s *set) Length() int { } func (s *set) Values() (values []string) { - for val, _ := range s.data { + for val := range s.data { values = append(values, val) } return diff --git a/vendor/github.com/coreos/go-systemd/dbus/subscription.go b/vendor/github.com/coreos/go-systemd/dbus/subscription.go index 996451445..70e63a6f1 100644 --- a/vendor/github.com/coreos/go-systemd/dbus/subscription.go +++ b/vendor/github.com/coreos/go-systemd/dbus/subscription.go @@ -16,6 +16,7 @@ package dbus import ( "errors" + "log" "time" "github.com/godbus/dbus" @@ -36,22 +37,12 @@ func (c *Conn) Subscribe() error { c.sigconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0, "type='signal',interface='org.freedesktop.DBus.Properties',member='PropertiesChanged'") - err := c.sigobj.Call("org.freedesktop.systemd1.Manager.Subscribe", 0).Store() - if err != nil { - return err - } - - return nil + return c.sigobj.Call("org.freedesktop.systemd1.Manager.Subscribe", 0).Store() } // Unsubscribe this connection from systemd dbus events. func (c *Conn) Unsubscribe() error { - err := c.sigobj.Call("org.freedesktop.systemd1.Manager.Unsubscribe", 0).Store() - if err != nil { - return err - } - - return nil + return c.sigobj.Call("org.freedesktop.systemd1.Manager.Unsubscribe", 0).Store() } func (c *Conn) dispatch() { @@ -70,7 +61,8 @@ func (c *Conn) dispatch() { c.jobComplete(signal) } - if c.subscriber.updateCh == nil { + if c.subStateSubscriber.updateCh == nil && + c.propertiesSubscriber.updateCh == nil { continue } @@ -84,6 +76,12 @@ func (c *Conn) dispatch() { case "org.freedesktop.DBus.Properties.PropertiesChanged": if signal.Body[0].(string) == "org.freedesktop.systemd1.Unit" { unitPath = signal.Path + + if len(signal.Body) >= 2 { + if changed, ok := signal.Body[1].(map[string]dbus.Variant); ok { + c.sendPropertiesUpdate(unitPath, changed) + } + } } } @@ -169,42 +167,80 @@ type SubStateUpdate struct { // is full, it attempts to write an error to errCh; if errCh is full, the error // passes silently. func (c *Conn) SetSubStateSubscriber(updateCh chan<- *SubStateUpdate, errCh chan<- error) { - c.subscriber.Lock() - defer c.subscriber.Unlock() - c.subscriber.updateCh = updateCh - c.subscriber.errCh = errCh + if c == nil { + msg := "nil receiver" + select { + case errCh <- errors.New(msg): + default: + log.Printf("full error channel while reporting: %s\n", msg) + } + return + } + + c.subStateSubscriber.Lock() + defer c.subStateSubscriber.Unlock() + c.subStateSubscriber.updateCh = updateCh + c.subStateSubscriber.errCh = errCh } -func (c *Conn) sendSubStateUpdate(path dbus.ObjectPath) { - c.subscriber.Lock() - defer c.subscriber.Unlock() +func (c *Conn) sendSubStateUpdate(unitPath dbus.ObjectPath) { + c.subStateSubscriber.Lock() + defer c.subStateSubscriber.Unlock() + + if c.subStateSubscriber.updateCh == nil { + return + } - if c.shouldIgnore(path) { + isIgnored := c.shouldIgnore(unitPath) + defer c.cleanIgnore() + if isIgnored { return } - info, err := c.GetUnitProperties(string(path)) + info, err := c.GetUnitPathProperties(unitPath) if err != nil { select { - case c.subscriber.errCh <- err: + case c.subStateSubscriber.errCh <- err: default: + log.Printf("full error channel while reporting: %s\n", err) } + return } + defer c.updateIgnore(unitPath, info) - name := info["Id"].(string) - substate := info["SubState"].(string) + name, ok := info["Id"].(string) + if !ok { + msg := "failed to cast info.Id" + select { + case c.subStateSubscriber.errCh <- errors.New(msg): + default: + log.Printf("full error channel while reporting: %s\n", err) + } + return + } + substate, ok := info["SubState"].(string) + if !ok { + msg := "failed to cast info.SubState" + select { + case c.subStateSubscriber.errCh <- errors.New(msg): + default: + log.Printf("full error channel while reporting: %s\n", msg) + } + return + } update := &SubStateUpdate{name, substate} select { - case c.subscriber.updateCh <- update: + case c.subStateSubscriber.updateCh <- update: default: + msg := "update channel is full" select { - case c.subscriber.errCh <- errors.New("update channel full!"): + case c.subStateSubscriber.errCh <- errors.New(msg): default: + log.Printf("full error channel while reporting: %s\n", msg) } + return } - - c.updateIgnore(path, info) } // The ignore functions work around a wart in the systemd dbus interface. @@ -222,29 +258,76 @@ func (c *Conn) sendSubStateUpdate(path dbus.ObjectPath) { // the properties). func (c *Conn) shouldIgnore(path dbus.ObjectPath) bool { - t, ok := c.subscriber.ignore[path] + t, ok := c.subStateSubscriber.ignore[path] return ok && t >= time.Now().UnixNano() } func (c *Conn) updateIgnore(path dbus.ObjectPath, info map[string]interface{}) { - c.cleanIgnore() + loadState, ok := info["LoadState"].(string) + if !ok { + return + } // unit is unloaded - it will trigger bad systemd dbus behavior - if info["LoadState"].(string) == "not-found" { - c.subscriber.ignore[path] = time.Now().UnixNano() + ignoreInterval + if loadState == "not-found" { + c.subStateSubscriber.ignore[path] = time.Now().UnixNano() + ignoreInterval } } // without this, ignore would grow unboundedly over time func (c *Conn) cleanIgnore() { now := time.Now().UnixNano() - if c.subscriber.cleanIgnore < now { - c.subscriber.cleanIgnore = now + cleanIgnoreInterval + if c.subStateSubscriber.cleanIgnore < now { + c.subStateSubscriber.cleanIgnore = now + cleanIgnoreInterval - for p, t := range c.subscriber.ignore { + for p, t := range c.subStateSubscriber.ignore { if t < now { - delete(c.subscriber.ignore, p) + delete(c.subStateSubscriber.ignore, p) } } } } + +// PropertiesUpdate holds a map of a unit's changed properties +type PropertiesUpdate struct { + UnitName string + Changed map[string]dbus.Variant +} + +// SetPropertiesSubscriber writes to updateCh when any unit's properties +// change. Every property change reported by systemd will be sent; that is, no +// transitions will be "missed" (as they might be with SetSubStateSubscriber). +// However, state changes will only be written to the channel with non-blocking +// writes. If updateCh is full, it attempts to write an error to errCh; if +// errCh is full, the error passes silently. +func (c *Conn) SetPropertiesSubscriber(updateCh chan<- *PropertiesUpdate, errCh chan<- error) { + c.propertiesSubscriber.Lock() + defer c.propertiesSubscriber.Unlock() + c.propertiesSubscriber.updateCh = updateCh + c.propertiesSubscriber.errCh = errCh +} + +// we don't need to worry about shouldIgnore() here because +// sendPropertiesUpdate doesn't call GetProperties() +func (c *Conn) sendPropertiesUpdate(unitPath dbus.ObjectPath, changedProps map[string]dbus.Variant) { + c.propertiesSubscriber.Lock() + defer c.propertiesSubscriber.Unlock() + + if c.propertiesSubscriber.updateCh == nil { + return + } + + update := &PropertiesUpdate{unitName(unitPath), changedProps} + + select { + case c.propertiesSubscriber.updateCh <- update: + default: + msg := "update channel is full" + select { + case c.propertiesSubscriber.errCh <- errors.New(msg): + default: + log.Printf("full error channel while reporting: %s\n", msg) + } + return + } +} diff --git a/vendor/github.com/coreos/go-systemd/journal/journal.go b/vendor/github.com/coreos/go-systemd/journal/journal.go index 7f434990d..ef85a3ba2 100644 --- a/vendor/github.com/coreos/go-systemd/journal/journal.go +++ b/vendor/github.com/coreos/go-systemd/journal/journal.go @@ -103,7 +103,10 @@ func Send(message string, priority Priority, vars map[string]string) error { if !ok { return journalError("can't send file through non-Unix connection") } - unixConn.WriteMsgUnix([]byte{}, rights, nil) + _, _, err = unixConn.WriteMsgUnix([]byte{}, rights, nil) + if err != nil { + return journalError(err.Error()) + } } else if err != nil { return journalError(err.Error()) } @@ -165,7 +168,7 @@ func tempFd() (*os.File, error) { if err != nil { return nil, err } - syscall.Unlink(file.Name()) + err = syscall.Unlink(file.Name()) if err != nil { return nil, err } diff --git a/vendor/github.com/coreos/go-systemd/sdjournal/journal.go b/vendor/github.com/coreos/go-systemd/sdjournal/journal.go index b00d606c1..9f3d92342 100644 --- a/vendor/github.com/coreos/go-systemd/sdjournal/journal.go +++ b/vendor/github.com/coreos/go-systemd/sdjournal/journal.go @@ -47,6 +47,15 @@ package sdjournal // return sd_journal_open_directory(ret, path, flags); // } // +// int +// my_sd_journal_open_files(void *f, sd_journal **ret, const char **paths, int flags) +// { +// int (*sd_journal_open_files)(sd_journal **, const char **, int); +// +// sd_journal_open_files = f; +// return sd_journal_open_files(ret, paths, flags); +// } +// // void // my_sd_journal_close(void *f, sd_journal *j) // { @@ -282,9 +291,19 @@ package sdjournal // sd_journal_restart_unique(j); // } // +// int +// my_sd_journal_get_catalog(void *f, sd_journal *j, char **ret) +// { +// int(*sd_journal_get_catalog)(sd_journal *, char **); +// +// sd_journal_get_catalog = f; +// return sd_journal_get_catalog(j, ret); +// } +// import "C" import ( "bytes" + "errors" "fmt" "strings" "sync" @@ -352,6 +371,12 @@ const ( IndefiniteWait time.Duration = 1<<63 - 1 ) +var ( + // ErrNoTestCursor gets returned when using TestCursor function and cursor + // parameter is not the same as the current cursor position. + ErrNoTestCursor = errors.New("Cursor parameter is not the same as current position") +) + // Journal is a Go wrapper of an sd_journal structure. type Journal struct { cjournal *C.sd_journal @@ -396,8 +421,7 @@ func NewJournal() (j *Journal, err error) { } // NewJournalFromDir returns a new Journal instance pointing to a journal residing -// in a given directory. The supplied path may be relative or absolute; if -// relative, it will be converted to an absolute path before being opened. +// in a given directory. func NewJournalFromDir(path string) (j *Journal, err error) { j = &Journal{} @@ -417,6 +441,32 @@ func NewJournalFromDir(path string) (j *Journal, err error) { return j, nil } +// NewJournalFromFiles returns a new Journal instance pointing to a journals residing +// in a given files. +func NewJournalFromFiles(paths ...string) (j *Journal, err error) { + j = &Journal{} + + sd_journal_open_files, err := getFunction("sd_journal_open_files") + if err != nil { + return nil, err + } + + // by making the slice 1 elem too long, we guarantee it'll be null-terminated + cPaths := make([]*C.char, len(paths)+1) + for idx, path := range paths { + p := C.CString(path) + cPaths[idx] = p + defer C.free(unsafe.Pointer(p)) + } + + r := C.my_sd_journal_open_files(sd_journal_open_files, &j.cjournal, &cPaths[0], 0) + if r < 0 { + return nil, fmt.Errorf("failed to open journals in paths %q: %d", paths, syscall.Errno(-r)) + } + + return j, nil +} + // Close closes a journal opened with NewJournal. func (j *Journal) Close() error { sd_journal_close, err := getFunction("sd_journal_close") @@ -598,7 +648,8 @@ func (j *Journal) getData(field string) (unsafe.Pointer, C.int, error) { } // GetData gets the data object associated with a specific field from the -// current journal entry. +// the journal entry referenced by the last completed Next/Previous function +// call. To call GetData, you must have first called one of these functions. func (j *Journal) GetData(field string) (string, error) { d, l, err := j.getData(field) if err != nil { @@ -609,7 +660,9 @@ func (j *Journal) GetData(field string) (string, error) { } // GetDataValue gets the data object associated with a specific field from the -// current journal entry, returning only the value of the object. +// journal entry referenced by the last completed Next/Previous function call, +// returning only the value of the object. To call GetDataValue, you must first +// have called one of the Next/Previous functions. func (j *Journal) GetDataValue(field string) (string, error) { val, err := j.GetData(field) if err != nil { @@ -620,7 +673,8 @@ func (j *Journal) GetDataValue(field string) (string, error) { } // GetDataBytes gets the data object associated with a specific field from the -// current journal entry. +// journal entry referenced by the last completed Next/Previous function call. +// To call GetDataBytes, you must first have called one of these functions. func (j *Journal) GetDataBytes(field string) ([]byte, error) { d, l, err := j.getData(field) if err != nil { @@ -631,7 +685,9 @@ func (j *Journal) GetDataBytes(field string) ([]byte, error) { } // GetDataValueBytes gets the data object associated with a specific field from the -// current journal entry, returning only the value of the object. +// journal entry referenced by the last completed Next/Previous function call, +// returning only the value of the object. To call GetDataValueBytes, you must first +// have called one of the Next/Previous functions. func (j *Journal) GetDataValueBytes(field string) ([]byte, error) { val, err := j.GetDataBytes(field) if err != nil { @@ -641,9 +697,10 @@ func (j *Journal) GetDataValueBytes(field string) ([]byte, error) { return bytes.SplitN(val, []byte("="), 2)[1], nil } -// GetEntry returns a full representation of a journal entry with -// all key-value pairs of data as well as address fields (cursor, realtime -// timestamp and monotonic timestamp) +// GetEntry returns a full representation of the journal entry referenced by the +// last completed Next/Previous function call, with all key-value pairs of data +// as well as address fields (cursor, realtime timestamp and monotonic timestamp). +// To call GetEntry, you must first have called one of the Next/Previous functions. func (j *Journal) GetEntry() (*JournalEntry, error) { sd_journal_get_realtime_usec, err := getFunction("sd_journal_get_realtime_usec") if err != nil { @@ -731,7 +788,7 @@ func (j *Journal) GetEntry() (*JournalEntry, error) { return entry, nil } -// SetDataThresold sets the data field size threshold for data returned by +// SetDataThreshold sets the data field size threshold for data returned by // GetData. To retrieve the complete data fields this threshold should be // turned off by setting it to 0, so that the library always returns the // complete data objects. @@ -752,8 +809,10 @@ func (j *Journal) SetDataThreshold(threshold uint64) error { return nil } -// GetRealtimeUsec gets the realtime (wallclock) timestamp of the current -// journal entry. +// GetRealtimeUsec gets the realtime (wallclock) timestamp of the journal +// entry referenced by the last completed Next/Previous function call. To +// call GetRealtimeUsec, you must first have called one of the Next/Previous +// functions. func (j *Journal) GetRealtimeUsec() (uint64, error) { var usec C.uint64_t @@ -773,7 +832,10 @@ func (j *Journal) GetRealtimeUsec() (uint64, error) { return uint64(usec), nil } -// GetMonotonicUsec gets the monotonic timestamp of the current journal entry. +// GetMonotonicUsec gets the monotonic timestamp of the journal entry +// referenced by the last completed Next/Previous function call. To call +// GetMonotonicUsec, you must first have called one of the Next/Previous +// functions. func (j *Journal) GetMonotonicUsec() (uint64, error) { var usec C.uint64_t var boot_id C.sd_id128_t @@ -794,7 +856,9 @@ func (j *Journal) GetMonotonicUsec() (uint64, error) { return uint64(usec), nil } -// GetCursor gets the cursor of the current journal entry. +// GetCursor gets the cursor of the last journal entry reeferenced by the +// last completed Next/Previous function call. To call GetCursor, you must +// first have called one of the Next/Previous functions. func (j *Journal) GetCursor() (string, error) { sd_journal_get_cursor, err := getFunction("sd_journal_get_cursor") if err != nil { @@ -836,13 +900,16 @@ func (j *Journal) TestCursor(cursor string) error { if r < 0 { return fmt.Errorf("failed to test to cursor %q: %d", cursor, syscall.Errno(-r)) + } else if r == 0 { + return ErrNoTestCursor } return nil } // SeekHead seeks to the beginning of the journal, i.e. the oldest available -// entry. +// entry. This call must be followed by a call to Next before any call to +// Get* will return data about the first element. func (j *Journal) SeekHead() error { sd_journal_seek_head, err := getFunction("sd_journal_seek_head") if err != nil { @@ -861,7 +928,8 @@ func (j *Journal) SeekHead() error { } // SeekTail may be used to seek to the end of the journal, i.e. the most recent -// available entry. +// available entry. This call must be followed by a call to Next before any +// call to Get* will return data about the last element. func (j *Journal) SeekTail() error { sd_journal_seek_tail, err := getFunction("sd_journal_seek_tail") if err != nil { @@ -880,7 +948,8 @@ func (j *Journal) SeekTail() error { } // SeekRealtimeUsec seeks to the entry with the specified realtime (wallclock) -// timestamp, i.e. CLOCK_REALTIME. +// timestamp, i.e. CLOCK_REALTIME. This call must be followed by a call to +// Next/Previous before any call to Get* will return data about the sought entry. func (j *Journal) SeekRealtimeUsec(usec uint64) error { sd_journal_seek_realtime_usec, err := getFunction("sd_journal_seek_realtime_usec") if err != nil { @@ -898,7 +967,9 @@ func (j *Journal) SeekRealtimeUsec(usec uint64) error { return nil } -// SeekCursor seeks to a concrete journal cursor. +// SeekCursor seeks to a concrete journal cursor. This call must be +// followed by a call to Next/Previous before any call to Get* will return +// data about the sought entry. func (j *Journal) SeekCursor(cursor string) error { sd_journal_seek_cursor, err := getFunction("sd_journal_seek_cursor") if err != nil { @@ -937,7 +1008,7 @@ func (j *Journal) Wait(timeout time.Duration) int { // equivalent hex value. to = 0xffffffffffffffff } else { - to = uint64(time.Now().Add(timeout).Unix() / 1000) + to = uint64(timeout / time.Microsecond) } j.mu.Lock() r := C.my_sd_journal_wait(sd_journal_wait, j.cjournal, C.uint64_t(to)) @@ -1022,3 +1093,28 @@ func (j *Journal) GetUniqueValues(field string) ([]string, error) { return result, nil } + +// GetCatalog retrieves a message catalog entry for the journal entry referenced +// by the last completed Next/Previous function call. To call GetCatalog, you +// must first have called one of these functions. +func (j *Journal) GetCatalog() (string, error) { + sd_journal_get_catalog, err := getFunction("sd_journal_get_catalog") + if err != nil { + return "", err + } + + var c *C.char + + j.mu.Lock() + r := C.my_sd_journal_get_catalog(sd_journal_get_catalog, j.cjournal, &c) + j.mu.Unlock() + defer C.free(unsafe.Pointer(c)) + + if r < 0 { + return "", fmt.Errorf("failed to retrieve catalog entry for current journal entry: %d", syscall.Errno(-r)) + } + + catalog := C.GoString(c) + + return catalog, nil +} diff --git a/vendor/github.com/coreos/go-systemd/sdjournal/read.go b/vendor/github.com/coreos/go-systemd/sdjournal/read.go index b581f03b4..51a060fb5 100644 --- a/vendor/github.com/coreos/go-systemd/sdjournal/read.go +++ b/vendor/github.com/coreos/go-systemd/sdjournal/read.go @@ -21,10 +21,13 @@ import ( "io" "log" "strings" + "sync" "time" ) var ( + // ErrExpired gets returned when the Follow function runs into the + // specified timeout. ErrExpired = errors.New("Timeout expired") ) @@ -44,6 +47,11 @@ type JournalReaderConfig struct { // If not empty, the journal instance will point to a journal residing // in this directory. The supplied path may be relative or absolute. Path string + + // If not nil, Formatter will be used to translate the resulting entries + // into strings. If not set, the default format (timestamp and message field) + // will be used. If Formatter returns an error, Read will stop and return the error. + Formatter func(entry *JournalEntry) (string, error) } // JournalReader is an io.ReadCloser which provides a simple interface for iterating through the @@ -51,12 +59,20 @@ type JournalReaderConfig struct { type JournalReader struct { journal *Journal msgReader *strings.Reader + formatter func(entry *JournalEntry) (string, error) } // NewJournalReader creates a new JournalReader with configuration options that are similar to the // systemd journalctl tool's iteration and filtering features. func NewJournalReader(config JournalReaderConfig) (*JournalReader, error) { - r := &JournalReader{} + // use simpleMessageFormatter as default formatter. + if config.Formatter == nil { + config.Formatter = simpleMessageFormatter + } + + r := &JournalReader{ + formatter: config.Formatter, + } // Open the journal var err error @@ -71,7 +87,9 @@ func NewJournalReader(config JournalReaderConfig) (*JournalReader, error) { // Add any supplied matches for _, m := range config.Matches { - r.journal.AddMatch(m.String()) + if err = r.journal.AddMatch(m.String()); err != nil { + return nil, err + } } // Set the start position based on options @@ -118,14 +136,10 @@ func NewJournalReader(config JournalReaderConfig) (*JournalReader, error) { // don't fit in the read buffer. Callers should keep calling until 0 and/or an // error is returned. func (r *JournalReader) Read(b []byte) (int, error) { - var err error - if r.msgReader == nil { - var c uint64 - // Advance the journal cursor. It has to be called at least one time // before reading - c, err = r.journal.Next() + c, err := r.journal.Next() // An unexpected error if err != nil { @@ -137,10 +151,13 @@ func (r *JournalReader) Read(b []byte) (int, error) { return 0, io.EOF } - // Build a message - var msg string - msg, err = r.buildMessage() + entry, err := r.journal.GetEntry() + if err != nil { + return 0, err + } + // Build a message + msg, err := r.formatter(entry) if err != nil { return 0, err } @@ -148,8 +165,7 @@ func (r *JournalReader) Read(b []byte) (int, error) { } // Copy and return the message - var sz int - sz, err = r.msgReader.Read(b) + sz, err := r.msgReader.Read(b) if err == io.EOF { // The current entry has been fully read. Don't propagate this // EOF, so the next entry can be read at the next Read() @@ -180,80 +196,76 @@ func (r *JournalReader) Rewind() error { // Follow synchronously follows the JournalReader, writing each new journal entry to writer. The // follow will continue until a single time.Time is received on the until channel. -func (r *JournalReader) Follow(until <-chan time.Time, writer io.Writer) (err error) { +func (r *JournalReader) Follow(until <-chan time.Time, writer io.Writer) error { // Process journal entries and events. Entries are flushed until the tail or // timeout is reached, and then we wait for new events or the timeout. var msg = make([]byte, 64*1<<(10)) + var waitCh = make(chan int, 1) + var waitGroup sync.WaitGroup + defer waitGroup.Wait() + process: for { c, err := r.Read(msg) if err != nil && err != io.EOF { - break process + return err } select { case <-until: return ErrExpired default: - if c > 0 { - if _, err = writer.Write(msg[:c]); err != nil { - break process - } - continue process + } + if c > 0 { + if _, err = writer.Write(msg[:c]); err != nil { + return err } + continue process } // We're at the tail, so wait for new events or time out. // Holds journal events to process. Tightly bounded for now unless there's a // reason to unblock the journal watch routine more quickly. - events := make(chan int, 1) - pollDone := make(chan bool, 1) - go func() { - for { - select { - case <-pollDone: - return + for { + waitGroup.Add(1) + go func() { + status := r.journal.Wait(100 * time.Millisecond) + waitCh <- status + waitGroup.Done() + }() + + select { + case <-until: + return ErrExpired + case e := <-waitCh: + switch e { + case SD_JOURNAL_NOP: + // the journal did not change since the last invocation + case SD_JOURNAL_APPEND, SD_JOURNAL_INVALIDATE: + continue process default: - events <- r.journal.Wait(time.Duration(1) * time.Second) - } - } - }() + if e < 0 { + return fmt.Errorf("received error event: %d", e) + } - select { - case <-until: - pollDone <- true - return ErrExpired - case e := <-events: - pollDone <- true - switch e { - case SD_JOURNAL_NOP, SD_JOURNAL_APPEND, SD_JOURNAL_INVALIDATE: - // TODO: need to account for any of these? - default: - log.Printf("Received unknown event: %d\n", e) + log.Printf("received unknown event: %d\n", e) + } } - continue process } } - - return } -// buildMessage returns a string representing the current journal entry in a simple format which +// simpleMessageFormatter is the default formatter. +// It returns a string representing the current journal entry in a simple format which // includes the entry timestamp and MESSAGE field. -func (r *JournalReader) buildMessage() (string, error) { - var msg string - var usec uint64 - var err error - - if msg, err = r.journal.GetData("MESSAGE"); err != nil { - return "", err - } - - if usec, err = r.journal.GetRealtimeUsec(); err != nil { - return "", err +func simpleMessageFormatter(entry *JournalEntry) (string, error) { + msg, ok := entry.Fields["MESSAGE"] + if !ok { + return "", fmt.Errorf("no MESSAGE field present in journal entry") } + usec := entry.RealtimeTimestamp timestamp := time.Unix(0, int64(usec)*int64(time.Microsecond)) return fmt.Sprintf("%s %s\n", timestamp, msg), nil -- cgit v1.2.3-54-g00ecf