Merge pull request #279 from lucab/ups/journal-connectionless

journal: use a connection-less socket

Author: Luca Bruno

.travis.yml

@@ -5,7 +5,6 @@ services:
 
 language: go
 go:
-  - "1.7.x"
   - "1.10.x"
   - "1.11.x"
 go_import_path: github.com/coreos/go-systemd
@@ -32,13 +31,13 @@ before_install:
  - rm -f /tmp/cidfile
 
 install:
- - docker run -d --cidfile=/tmp/cidfile --privileged -e GOPATH=${GOPATH} -v ${PWD}:${BUILD_DIR} go-systemd/container-tests /bin/systemd --system
+ - docker run --shm-size=2gb -d --cidfile=/tmp/cidfile --privileged -e GOPATH=${GOPATH} -v ${PWD}:${BUILD_DIR} go-systemd/container-tests /bin/systemd --system
 
 script:
  - ./scripts/travis/pr-test.sh go_fmt
  - ./scripts/travis/pr-test.sh build_source
  - ./scripts/travis/pr-test.sh build_tests
- - docker exec `cat /tmp/cidfile` /bin/bash -c "cd ${BUILD_DIR} && ./scripts/travis/pr-test.sh run_tests"
+ - docker exec --privileged `cat /tmp/cidfile` /bin/bash -c "cd ${BUILD_DIR} && ./scripts/travis/pr-test.sh run_tests"
  - ./scripts/travis/pr-test.sh go_vet
  - ./scripts/travis/pr-test.sh license_check
 

README.md

@@ -2,7 +2,7 @@
 
 [![Build Status](https://travis-ci.org/coreos/go-systemd.png?branch=master)](https://travis-ci.org/coreos/go-systemd)
 [![godoc](https://godoc.org/github.com/coreos/go-systemd?status.svg)](http://godoc.org/github.com/coreos/go-systemd)
-![minimum golang 1.7](https://img.shields.io/badge/golang-1.7%2B-orange.svg)
+![minimum golang 1.10](https://img.shields.io/badge/golang-1.10%2B-orange.svg)
 
 
 Go bindings to systemd. The project has several packages:

journal/journal.go

@@ -33,7 +33,10 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"syscall"
+	"unsafe"
 )
 
 // Priority of a journal message
@@ -50,19 +53,35 @@ const (
 	PriDebug
 )
 
-var conn net.Conn
+var (
+	// This can be overridden at build-time:
+	// https://github.com/golang/go/wiki/GcToolchainTricks#including-build-information-in-the-executable
+	journalSocket = "/run/systemd/journal/socket"
+
+	// unixConnPtr atomically holds the local unconnected Unix-domain socket.
+	// Concrete safe pointer type: *net.UnixConn
+	unixConnPtr unsafe.Pointer
+	// onceConn ensures that unixConnPtr is initialized exactly once.
+	onceConn sync.Once
+)
 
 func init() {
-	var err error
-	conn, err = net.Dial("unixgram", "/run/systemd/journal/socket")
-	if err != nil {
-		conn = nil
-	}
+	onceConn.Do(initConn)
 }
 
-// Enabled returns true if the local systemd journal is available for logging
+// Enabled checks whether the local systemd journal is available for logging.
 func Enabled() bool {
-	return conn != nil
+	onceConn.Do(initConn)
+
+	if (*net.UnixConn)(atomic.LoadPointer(&unixConnPtr)) == nil {
+		return false
+	}
+
+	if _, err := net.Dial("unixgram", journalSocket); err != nil {
+		return false
+	}
+
+	return true
 }
 
 // Send a message to the local systemd journal. vars is a map of journald
@@ -73,8 +92,14 @@ func Enabled() bool {
 // (http://www.freedesktop.org/software/systemd/man/systemd.journal-fields.html)
 // for more details.  vars may be nil.
 func Send(message string, priority Priority, vars map[string]string) error {
+	conn := (*net.UnixConn)(atomic.LoadPointer(&unixConnPtr))
 	if conn == nil {
-		return journalError("could not connect to journald socket")
+		return errors.New("could not initialize socket to journald")
+	}
+
+	socketAddr := &net.UnixAddr{
+		Name: journalSocket,
+		Net:  "unixgram",
 	}
 
 	data := new(bytes.Buffer)
@@ -84,32 +109,30 @@ func Send(message string, priority Priority, vars map[string]string) error {
 		appendVariable(data, k, v)
 	}
 
-	_, err := io.Copy(conn, data)
-	if err != nil && isSocketSpaceError(err) {
-		file, err := tempFd()
-		if err != nil {
-			return journalError(err.Error())
-		}
-		defer file.Close()
-		_, err = io.Copy(file, data)
-		if err != nil {
-			return journalError(err.Error())
-		}
-
-		rights := syscall.UnixRights(int(file.Fd()))
+	_, _, err := conn.WriteMsgUnix(data.Bytes(), nil, socketAddr)
+	if err == nil {
+		return nil
+	}
+	if !isSocketSpaceError(err) {
+		return err
+	}
 
-		/* this connection should always be a UnixConn, but better safe than sorry */
-		unixConn, ok := conn.(*net.UnixConn)
-		if !ok {
-			return journalError("can't send file through non-Unix connection")
-		}
-		_, _, err = unixConn.WriteMsgUnix([]byte{}, rights, nil)
-		if err != nil {
-			return journalError(err.Error())
-		}
-	} else if err != nil {
-		return journalError(err.Error())
+	// Large log entry, send it via tempfile and ancillary-fd.
+	file, err := tempFd()
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+	_, err = io.Copy(file, data)
+	if err != nil {
+		return err
+	}
+	rights := syscall.UnixRights(int(file.Fd()))
+	_, _, err = conn.WriteMsgUnix([]byte{}, rights, socketAddr)
+	if err != nil {
+		return err
 	}
+
 	return nil
 }
 
@@ -120,7 +143,7 @@ func Print(priority Priority, format string, a ...interface{}) error {
 
 func appendVariable(w io.Writer, name, value string) {
 	if err := validVarName(name); err != nil {
-		journalError(err.Error())
+		fmt.Fprintf(os.Stderr, "variable name %s contains invalid character, ignoring\n", name)
 	}
 	if strings.ContainsRune(value, '\n') {
 		/* When the value contains a newline, we write:
@@ -137,9 +160,9 @@ func appendVariable(w io.Writer, name, value string) {
 	}
 }
 
-// validVarName validates a variable name to make sure it journald will accept it.
+// validVarName validates a variable name to make sure journald will accept it.
 // The variable name must be in uppercase and consist only of characters,
-// numbers and underscores, and may not begin with an underscore. (from the docs)
+// numbers and underscores, and may not begin with an underscore:
 // https://www.freedesktop.org/software/systemd/man/sd_journal_print.html
 func validVarName(name string) error {
 	if name == "" {
@@ -156,20 +179,23 @@ func validVarName(name string) error {
 	return nil
 }
 
+// isSocketSpaceError checks whether the error is signaling
+// an "overlarge message" condition.
 func isSocketSpaceError(err error) bool {
 	opErr, ok := err.(*net.OpError)
-	if !ok {
+	if !ok || opErr == nil {
 		return false
 	}
 
-	sysErr, ok := opErr.Err.(syscall.Errno)
-	if !ok {
+	sysErr, ok := opErr.Err.(*os.SyscallError)
+	if !ok || sysErr == nil {
 		return false
 	}
 
-	return sysErr == syscall.EMSGSIZE || sysErr == syscall.ENOBUFS
+	return sysErr.Err == syscall.EMSGSIZE || sysErr.Err == syscall.ENOBUFS
 }
 
+// tempFd creates a temporary, unlinked file under `/dev/shm`.
 func tempFd() (*os.File, error) {
 	file, err := ioutil.TempFile("/dev/shm/", "journal.XXXXX")
 	if err != nil {
@@ -182,8 +208,18 @@ func tempFd() (*os.File, error) {
 	return file, nil
 }
 
-func journalError(s string) error {
-	s = "journal error: " + s
-	fmt.Fprintln(os.Stderr, s)
-	return errors.New(s)
+// initConn initializes the global `unixConnPtr` socket.
+// It is meant to be called exactly once, at program startup.
+func initConn() {
+	autobind, err := net.ResolveUnixAddr("unixgram", "")
+	if err != nil {
+		return
+	}
+
+	sock, err := net.ListenUnixgram("unixgram", autobind)
+	if err != nil {
+		return
+	}
+
+	atomic.StorePointer(&unixConnPtr, unsafe.Pointer(sock))
 }

journal/journal_test.go

@@ -17,11 +17,20 @@ package journal
 import (
 	"fmt"
 	"io/ioutil"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
 )
 
+func TestJournalEnabled(t *testing.T) {
+	enabled := Enabled()
+
+	if !enabled {
+		t.Fatalf("journald socket not detected")
+	}
+}
+
 func TestValidVarName(t *testing.T) {
 	validTestCases := []string{
 		"TEST",
@@ -65,8 +74,6 @@ func TestJournalSend(t *testing.T) {
 			largeValue = v + 1
 		}
 	}
-	// See https://github.com/coreos/go-systemd/pull/221#issuecomment-276727718
-	_ = largeValue
 
 	// small messages should go over normal data,
 	// larger ones over temporary file with fd in ancillary data
@@ -82,7 +89,6 @@ func TestJournalSend(t *testing.T) {
 			"small message",
 			5,
 		},
-		/* See https://github.com/coreos/go-systemd/pull/221#issuecomment-276727718
 		{
 			"large message",
 			largeValue,
@@ -91,18 +97,25 @@ func TestJournalSend(t *testing.T) {
 			"huge message",
 			hugeValue,
 		},
-		*/
 	}
 
+	// This is memory intensive, so we manually trigger GC before and after each test.
 	for i, tt := range testValues {
 		t.Logf("journal send test #%v - %s (len=%d)", i, tt.label, tt.len)
-		largeVars := map[string]string{
-			"KEY": string(make([]byte, tt.len)),
-		}
-
-		err := Send(fmt.Sprintf("go-systemd test #%v - %s", i, tt.label), PriCrit, largeVars)
+		runtime.GC()
+		err := SendAlloc(i, tt.label, tt.len)
 		if err != nil {
 			t.Fatalf("#%v: failed sending %s: %s", i, tt.label, err)
 		}
+		runtime.GC()
 	}
 }
+
+func SendAlloc(run int, label string, len int) error {
+	largeVars := map[string]string{
+		"KEY": string(make([]byte, len)),
+	}
+
+	msg := fmt.Sprintf("go-systemd test #%v - %s", run, label)
+	return Send(msg, PriCrit, largeVars)
+}