Alpha logging support (JSON+mirroring)

cmd/runmqserver/crtmqvol.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -16,11 +16,12 @@ limitations under the License.
 package main
 
 import (
-	"log"
 	"os"
 	"path/filepath"
 	"runtime"
 	"syscall"
+
+	log "github.com/sirupsen/logrus"
 )
 
 const mqmUID uint32 = 999

cmd/runmqserver/license.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -18,10 +18,11 @@ package main
 import (
 	"errors"
 	"io/ioutil"
-	"log"
 	"os"
 	"path/filepath"
 	"strings"
+
+	log "github.com/sirupsen/logrus"
 )
 
 // resolveLicenseFile returns the file name of the MQ license file, taking into

cmd/runmqserver/logging.go

@@ -0,0 +1,182 @@
+/*
+© Copyright IBM Corporation 2018
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// waitForFile waits until the specified file exists
+func waitForFile(path string) (os.FileInfo, error) {
+	var fi os.FileInfo
+	var err error
+	// Wait for file to exist
+	for {
+		fi, err = os.Stat(path)
+		if err != nil {
+			if os.IsNotExist(err) {
+				time.Sleep(500 * time.Millisecond)
+				continue
+			} else {
+				return nil, err
+			}
+		}
+		break
+	}
+	log.Debugf("File exists: %v, %v", path, fi.Size())
+	return fi, nil
+}
+
+// mirrorAvailableMessages prints lines from the file, until no more are available
+func mirrorAvailableMessages(f *os.File, w io.Writer) {
+	scanner := bufio.NewScanner(f)
+	count := 0
+	for scanner.Scan() {
+		t := scanner.Text()
+		if strings.HasPrefix(t, "{") {
+			// Assume JSON, so just print it
+			fmt.Fprintln(w, t)
+		} else if strings.HasPrefix(t, "AMQ") {
+			// Only print MQ messages with AMQnnnn codes
+			fmt.Fprintln(w, t)
+		}
+		count++
+	}
+	log.Debugf("Mirrored %v log entries", count)
+	err := scanner.Err()
+	if err != nil {
+		log.Errorf("Error reading file: %v", err)
+		return
+	}
+}
+
+// mirrorLog tails the specified file, and logs each line to stdout.
+// This is useful for usability, as the container console log can show
+// messages from the MQ error logs.
+func mirrorLog(path string, w io.Writer) (chan bool, error) {
+	lifecycle := make(chan bool)
+	var offset int64 = -1
+	var f *os.File
+	var err error
+	var fi os.FileInfo
+	// Need to check if the file exists before returning, otherwise we have a
+	// race to see if the new file get created before we can test for it
+	fi, err = os.Stat(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			// File doesn't exist, so ensure we start at the beginning
+			offset = 0
+		} else {
+			return nil, err
+		}
+	} else {
+		// If the file exists, open it now, before we return.  This makes sure
+		// the file is open before the queue manager is created or started.
+		// Otherwise, there would be the potential for a nearly-full file to
+		// rotate before the goroutine had a chance to open it.
+		f, err = os.OpenFile(path, os.O_RDONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		// File already exists, so start reading at the end
+		offset = fi.Size()
+	}
+
+	go func() {
+		if f == nil {
+			// File didn't exist, so need to wait for it
+			fi, err = waitForFile(path)
+			if err != nil {
+				log.Errorln(err)
+				lifecycle <- true
+				return
+			}
+			f, err = os.OpenFile(path, os.O_RDONLY, 0)
+			if err != nil {
+				log.Errorln(err)
+				lifecycle <- true
+				return
+			}
+		}
+
+		fi, err = f.Stat()
+		if err != nil {
+			log.Errorln(err)
+			lifecycle <- true
+			return
+		}
+		// The file now exists.  If it didn't exist before we started, offset=0
+		if offset != 0 {
+			log.Debugf("Seeking %v", offset)
+			f.Seek(offset, 0)
+		}
+		closing := false
+		for {
+			log.Debugln("Start of loop")
+			// If there's already data there, mirror it now.
+			mirrorAvailableMessages(f, w)
+			log.Debugf("Stat %v", path)
+			newFI, err := os.Stat(path)
+			if err != nil {
+				log.Error(err)
+				lifecycle <- true
+				return
+			}
+			if !os.SameFile(fi, newFI) {
+				log.Debugln("Not the same file!")
+				// WARNING: There is a possible race condition here.  If *another*
+				// log rotation happens before we can open the new file, then we
+				// could skip all those messages.  This could happen with a very small
+				// MQ error log size.
+				mirrorAvailableMessages(f, w)
+				f.Close()
+				// Re-open file
+				log.Debugln("Re-opening error log file")
+				// Used to work with this: f, err = waitForFile2(path)
+				f, err = os.OpenFile(path, os.O_RDONLY, 0)
+				if err != nil {
+					fmt.Printf("ERROR: %v", err)
+					return
+				}
+				fi = newFI
+				// Don't seek this time, because we know it's a new file
+				mirrorAvailableMessages(f, w)
+			}
+			log.Debugln("Check for lifecycle event")
+			select {
+			// Have we been asked to shut down?
+			case <-lifecycle:
+				// Set a flag, to allow one more time through the loop
+				closing = true
+			default:
+				if closing {
+					log.Debugln("Shutting down mirror")
+					lifecycle <- true
+					return
+				}
+				time.Sleep(500 * time.Millisecond)
+			}
+		}
+	}()
+	return lifecycle, nil
+}

cmd/runmqserver/logging_test.go

@@ -0,0 +1,149 @@
+/*
+© Copyright IBM Corporation 2018
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func TestMirrorLogWithoutRotation(t *testing.T) {
+	// Repeat the test multiple times, to help identify timing problems
+	for i := 0; i < 10; i++ {
+		t.Run(t.Name()+strconv.Itoa(i), func(t *testing.T) {
+			// Use just the sub-test name in the file name
+			tmp, err := ioutil.TempFile("", strings.Split(t.Name(), "/")[1])
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Log(tmp.Name())
+			defer os.Remove(tmp.Name())
+			b := make([]byte, 256)
+			buf := bytes.NewBuffer(b)
+			lifecycle, err := mirrorLog(tmp.Name(), buf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			f, err := os.OpenFile(tmp.Name(), os.O_WRONLY, 0700)
+			if err != nil {
+				t.Fatal(err)
+			}
+			log.Println("Logging 3 JSON messages")
+			fmt.Fprintln(f, "{\"message\"=\"A\"}")
+			fmt.Fprintln(f, "{\"message\"=\"B\"}")
+			fmt.Fprintln(f, "{\"message\"=\"C\"}")
+			f.Close()
+			lifecycle <- true
+			<-lifecycle
+
+			// Read the buffer back and count the lines
+			r := strings.NewReader(buf.String())
+			scanner := bufio.NewScanner(r)
+			count := 0
+			for scanner.Scan() {
+				count++
+				t.Logf("Received entry: %v", scanner.Text())
+			}
+			err = scanner.Err()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if count != 3 {
+				t.Fatalf("Expected 3 log entries; got %v", count)
+			}
+		})
+	}
+}
+
+func TestMirrorLogWithRotation(t *testing.T) {
+	// Repeat the test multiple times, to help identify timing problems
+	for i := 0; i < 5; i++ {
+		t.Run(t.Name()+strconv.Itoa(i), func(t *testing.T) {
+			// Use just the sub-test name in the file name
+			tmp, err := ioutil.TempFile("", strings.Split(t.Name(), "/")[1])
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Log(tmp.Name())
+			defer func() {
+				t.Log("Removing file")
+				os.Remove(tmp.Name())
+			}()
+			b := make([]byte, 512)
+			buf := bytes.NewBuffer(b)
+			lifecycle, err := mirrorLog(tmp.Name(), buf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			f, err := os.OpenFile(tmp.Name(), os.O_WRONLY, 0700)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Log("Logging 3 JSON messages")
+			fmt.Fprintln(f, "{\"message\"=\"A\"}")
+			fmt.Fprintln(f, "{\"message\"=\"B\"}")
+			fmt.Fprintln(f, "{\"message\"=\"C\"}")
+			f.Close()
+
+			// Rotate the file, by renaming it
+			rotated := tmp.Name() + ".1"
+			os.Rename(tmp.Name(), rotated)
+			defer os.Remove(rotated)
+			// Open a new file, with the same name as before
+			f, err = os.OpenFile(tmp.Name(), os.O_WRONLY|os.O_CREATE, 0700)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Log("Logging 2 more JSON messages")
+			fmt.Fprintln(f, "{\"message\"=\"D\"}")
+			fmt.Fprintln(f, "{\"message\"=\"E\"}")
+			f.Close()
+
+			// Shut the mirroring down
+			lifecycle <- true
+			// Wait until it's finished
+			<-lifecycle
+
+			// Read the buffer back and count the lines
+			r := strings.NewReader(buf.String())
+			scanner := bufio.NewScanner(r)
+			count := 0
+			for scanner.Scan() {
+				count++
+				t.Logf("Received entry: %v", scanner.Text())
+			}
+			err = scanner.Err()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if count != 5 {
+				t.Fatalf("Expected 5 log entries; got %v", count)
+			}
+		})
+	}
+}
+
+func init() {
+	log.SetLevel(log.DebugLevel)
+}

cmd/runmqserver/main.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -19,14 +19,15 @@ package main
 
 import (
 	"errors"
-	"fmt"
 	"io/ioutil"
-	"log"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
 
+	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
+
 	"github.com/ibm-messaging/mq-container/internal/command"
 	"github.com/ibm-messaging/mq-container/internal/name"
 )
@@ -35,13 +36,13 @@ var debug = false
 
 func logDebug(msg string) {
 	if debug {
-		log.Printf("DEBUG: %v", msg)
+		log.Debug(msg)
 	}
 }
 
 func logDebugf(format string, args ...interface{}) {
 	if debug {
-		log.Printf("DEBUG: %v", fmt.Sprintf(format, args...))
+		log.Debugf(format, args...)
 	}
 }
 
@@ -74,6 +75,7 @@ func createQueueManager(name string) error {
 func updateCommandLevel() error {
 	level, ok := os.LookupEnv("MQ_CMDLEVEL")
 	if ok && level != "" {
+		log.Printf("Setting CMDLEVEL to %v", level)
 		out, rc, err := command.Run("strmqm", "-e", "CMDLEVEL="+level)
 		if err != nil {
 			log.Printf("Error %v setting CMDLEVEL: %v", rc, string(out))
@@ -141,11 +143,56 @@ func stopQueueManager(name string) error {
 	return nil
 }
 
+func jsonLogs() bool {
+	e := os.Getenv("MQ_ALPHA_JSON_LOGS")
+	if e == "true" || e == "1" {
+		return true
+	}
+	return false
+}
+
+func mirrorLogs() bool {
+	e := os.Getenv("MQ_ALPHA_MIRROR_ERROR_LOGS")
+	if e == "true" || e == "1" {
+		return true
+	}
+	return false
+}
+
+func configureLogger(name string) {
+	if jsonLogs() {
+		formatter := logrus.JSONFormatter{
+			FieldMap: logrus.FieldMap{
+				logrus.FieldKeyMsg:   "message",
+				logrus.FieldKeyLevel: "ibm_level",
+				logrus.FieldKeyTime:  "ibm_datetime",
+			},
+			// Match time stamp format used by MQ messages (includes milliseconds)
+			TimestampFormat: "2006-01-02T15:04:05.000Z07:00",
+		}
+		logrus.SetFormatter(&formatter)
+	} else {
+		formatter := logrus.TextFormatter{
+			FullTimestamp: true,
+		}
+		logrus.SetFormatter(&formatter)
+	}
+	if debug {
+		logrus.SetLevel(logrus.DebugLevel)
+	}
+}
+
 func doMain() error {
 	debugEnv, ok := os.LookupEnv("DEBUG")
 	if ok && (debugEnv == "true" || debugEnv == "1") {
 		debug = true
 	}
+	name, err := name.GetQueueManagerName()
+	if err != nil {
+		log.Println(err)
+		return err
+	}
+	configureLogger(name)
 	accepted, err := checkLicense()
 	if err != nil {
 		return err
@@ -153,12 +200,6 @@ func doMain() error {
 	if !accepted {
 		return errors.New("License not accepted")
 	}
-
-	name, err := name.GetQueueManagerName()
-	if err != nil {
-		log.Println(err)
-		return err
-	}
 	log.Printf("Using queue manager name: %v", name)
 
 	// Start signal handler
@@ -174,6 +215,19 @@ func doMain() error {
 	if err != nil {
 		return err
 	}
+	var mirrorLifecycle chan bool
+	if mirrorLogs() {
+		f := "/var/mqm/qmgrs/" + name + "/errors/AMQERR01"
+		if jsonLogs() {
+			f = f + ".json"
+		} else {
+			f = f + ".LOG"
+		}
+		mirrorLifecycle, err = mirrorLog(f, os.Stdout)
+		if err != nil {
+			return err
+		}
+	}
 	err = createQueueManager(name)
 	if err != nil {
 		return err
@@ -195,6 +249,12 @@ func doMain() error {
 	signalControl <- reapNow
 	// Wait for terminate signal
 	<-signalControl
+	if mirrorLogs() {
+		// Tell the mirroring goroutine to shutdown
+		mirrorLifecycle <- true
+		// Wait for the mirroring goroutine to finish cleanly
+		<-mirrorLifecycle
+	}
 	return nil
 }
 

cmd/runmqserver/main_test.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -18,9 +18,10 @@ package main
 import (
 	"flag"
 	"io/ioutil"
-	"log"
 	"strconv"
 	"testing"
+
+	log "github.com/sirupsen/logrus"
 )
 
 var test *bool

cmd/runmqserver/mqconfig.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -17,12 +17,12 @@ package main
 
 import (
 	"io/ioutil"
-	"log"
 	"os/user"
 	"runtime"
 	"strings"
 
 	"github.com/ibm-messaging/mq-container/internal/capabilities"
+	log "github.com/sirupsen/logrus"
 )
 
 func logBaseImage() error {

cmd/runmqserver/mqconfig_linux.go

@@ -1,5 +1,20 @@
 // +build linux
 
+/*
+© Copyright IBM Corporation 2017, 2018
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
 package main
 
 import (

cmd/runmqserver/mqconfig_other.go

@@ -1,5 +1,20 @@
 // +build !linux
 
+/*
+© Copyright IBM Corporation 2018
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
 package main
 
 // Dummy version of this function, only for non-Linux systems.

cmd/runmqserver/signals.go

@@ -1,5 +1,5 @@
 /*
-© Copyright IBM Corporation 2017
+© Copyright IBM Corporation 2017, 2018
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@ limitations under the License.
 package main
 
 import (
-	"log"
 	"os"
 	"os/signal"
 	"syscall"
 
+	log "github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
 )