HTRACE-89. htraced: add log levels, writing to log files (cmccabe)
diff --git a/htrace-core/src/go/src/org/apache/htrace/common/log.go b/htrace-core/src/go/src/org/apache/htrace/common/log.go
new file mode 100644
index 0000000..31faea4
--- /dev/null
+++ b/htrace-core/src/go/src/org/apache/htrace/common/log.go
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 common
+
+import (
+ "errors"
+ "fmt"
+ "org/apache/htrace/conf"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+)
+
+// A logSink is a place logs can be written to.
+type logSink struct {
+ path logPath
+ file *os.File
+ lock sync.Mutex
+ refCount int // protected by logFilesLock
+}
+
+// Write to the logSink.
+func (sink *logSink) write(str string) {
+ sink.lock.Lock()
+ defer sink.lock.Unlock()
+ _, err := sink.file.Write([]byte(str))
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Error logging to '%s': %s\n", sink.path, err.Error())
+ }
+}
+
+// Unreference the logSink. If there are no more references, and the logSink is
+// closeable, then we will close it here.
+func (sink *logSink) Unref() {
+ logFilesLock.Lock()
+ defer logFilesLock.Unlock()
+ sink.refCount--
+ if sink.refCount <= 0 {
+ if sink.path.IsCloseable() {
+ err := sink.file.Close()
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Error closing log file %s: %s\n",
+ sink.path, err.Error())
+ }
+ }
+ logSinks[sink.path] = nil
+ }
+}
+
+type logPath string
+
+// An empty LogPath represents "stdout."
+const STDOUT_LOG_PATH = ""
+
+// Convert a path to a logPath.
+func logPathFromString(path string) logPath {
+ if path == STDOUT_LOG_PATH {
+ return logPath("")
+ }
+ absPath, err := filepath.Abs(path)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Failed to get absolute path of %s: %s\n",
+ path, err.Error())
+ return logPath(path)
+ }
+ return logPath(absPath)
+}
+
+// Convert the path to a human-readable string.
+func (path logPath) String() string {
+ if path == "" {
+ return "(stdout)"
+ } else {
+ return string(path)
+ }
+}
+
+// Return true if the path is closeable. stdout is not closeable.
+func (path logPath) IsCloseable() bool {
+ return path != STDOUT_LOG_PATH
+}
+
+func (path logPath) Open() *logSink {
+ if path == STDOUT_LOG_PATH {
+ return &logSink{path: path, file: os.Stdout}
+ }
+ file, err := os.OpenFile(string(path), os.O_WRONLY|os.O_APPEND, 0777)
+ if err != nil {
+ sink := &logSink{path: STDOUT_LOG_PATH, file: os.Stdout}
+ fmt.Fprintf(os.Stderr, "Failed to open log file %s: %s\n",
+ path, err.Error())
+ return sink
+ }
+ return &logSink{path: path, file: file}
+}
+
+var logFilesLock sync.Mutex
+
+var logSinks map[logPath]*logSink = make(map[logPath]*logSink)
+
+func getOrCreateLogSink(pathStr string) *logSink {
+ path := logPathFromString(pathStr)
+ logFilesLock.Lock()
+ defer logFilesLock.Unlock()
+ sink := logSinks[path]
+ if sink == nil {
+ sink = path.Open()
+ logSinks[path] = sink
+ }
+ sink.refCount++
+ return sink
+}
+
+type Level int
+
+const (
+ TRACE Level = iota
+ DEBUG
+ INFO
+ WARN
+ ERROR
+)
+
+var levelToString map[Level]string = map[Level]string{
+ TRACE: "TRACE",
+ DEBUG: "DEBUG",
+ INFO: "INFO",
+ WARN: "WARN",
+ ERROR: "ERROR",
+}
+
+func (level Level) String() string {
+ return levelToString[level]
+}
+
+func (level Level) LogString() string {
+ return level.String()[0:1]
+}
+
+func LevelFromString(str string) (Level, error) {
+ for k, v := range levelToString {
+ if strings.ToLower(v) == strings.ToLower(str) {
+ return k, nil
+ }
+ }
+ var levelNames sort.StringSlice
+ levelNames = make([]string, len(levelToString))
+ var i int
+ for _, v := range levelToString {
+ levelNames[i] = v
+ i++
+ }
+ sort.Sort(levelNames)
+ return TRACE, errors.New(fmt.Sprintf("No such level as '%s'. Valid "+
+ "levels are '%v'\n", str, levelNames))
+}
+
+type Logger struct {
+ sink *logSink
+ level Level
+}
+
+func NewLogger(faculty string, cnf *conf.Config) *Logger {
+ path, level := parseConf(faculty, cnf)
+ sink := getOrCreateLogSink(path)
+ return &Logger{sink: sink, level: level}
+}
+
+func parseConf(faculty string, cnf *conf.Config) (string, Level) {
+ facultyLogPathKey := faculty + "." + conf.HTRACE_LOG_PATH
+ var facultyLogPath string
+ if cnf.Contains(facultyLogPathKey) {
+ facultyLogPath = cnf.Get(facultyLogPathKey)
+ } else {
+ facultyLogPath = cnf.Get(conf.HTRACE_LOG_PATH)
+ }
+ facultyLogLevelKey := faculty + conf.HTRACE_LOG_LEVEL
+ var facultyLogLevelStr string
+ if cnf.Contains(facultyLogLevelKey) {
+ facultyLogLevelStr = cnf.Get(facultyLogLevelKey)
+ } else {
+ facultyLogLevelStr = cnf.Get(conf.HTRACE_LOG_LEVEL)
+ }
+ level, err := LevelFromString(facultyLogLevelStr)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "Error configuring log level: %s. Using TRACE.\n")
+ level = TRACE
+ }
+ return facultyLogPath, level
+}
+
+func (lg *Logger) Trace(str string) {
+ lg.write(TRACE, str)
+}
+
+func (lg *Logger) Tracef(format string, v ...interface{}) {
+ lg.write(TRACE, fmt.Sprintf(format, v...))
+}
+
+func (lg *Logger) Debug(str string) {
+ lg.write(DEBUG, str)
+}
+
+func (lg *Logger) Debugf(format string, v ...interface{}) {
+ lg.write(DEBUG, fmt.Sprintf(format, v...))
+}
+
+func (lg *Logger) Info(str string) {
+ lg.write(INFO, str)
+}
+
+func (lg *Logger) Infof(format string, v ...interface{}) {
+ lg.write(INFO, fmt.Sprintf(format, v...))
+}
+
+func (lg *Logger) Warn(str string) error {
+ lg.write(WARN, str)
+ return errors.New(str)
+}
+
+func (lg *Logger) Warnf(format string, v ...interface{}) error {
+ str := fmt.Sprintf(format, v...)
+ lg.write(WARN, str)
+ return errors.New(str)
+}
+
+func (lg *Logger) Error(str string) error {
+ lg.write(ERROR, str)
+ return errors.New(str)
+}
+
+func (lg *Logger) Errorf(format string, v ...interface{}) error {
+ str := fmt.Sprintf(format, v...)
+ lg.write(ERROR, str)
+ return errors.New(str)
+}
+
+func (lg *Logger) write(level Level, str string) {
+ if level >= lg.level {
+ lg.sink.write(time.Now().Format(time.RFC3339) + " " +
+ level.LogString() + ": " + str)
+ }
+}
+
+func (lg *Logger) Close() {
+ lg.sink.Unref()
+ lg.sink = nil
+}
diff --git a/htrace-core/src/go/src/org/apache/htrace/conf/config.go b/htrace-core/src/go/src/org/apache/htrace/conf/config.go
index 41e39fa..d905322 100644
--- a/htrace-core/src/go/src/org/apache/htrace/conf/config.go
+++ b/htrace-core/src/go/src/org/apache/htrace/conf/config.go
@@ -152,6 +152,12 @@
return &cnf, nil
}
+// Returns true if the configuration has a non-default value for the given key.
+func (cnf *Config) Contains(key string) bool {
+ _, ok := cnf.settings[key]
+ return ok
+}
+
// Get a string configuration key.
func (cnf *Config) Get(key string) string {
ret := cnf.settings[key]
diff --git a/htrace-core/src/go/src/org/apache/htrace/conf/config_keys.go b/htrace-core/src/go/src/org/apache/htrace/conf/config_keys.go
index b4e5994..5e359f7 100644
--- a/htrace-core/src/go/src/org/apache/htrace/conf/config_keys.go
+++ b/htrace-core/src/go/src/org/apache/htrace/conf/config_keys.go
@@ -52,6 +52,12 @@
// How many writes to buffer before applying backpressure to span senders.
const HTRACE_DATA_STORE_SPAN_BUFFER_SIZE = "data.store.span.buffer.size"
+// Path to put the logs from htrace, or the empty string to use stdout.
+const HTRACE_LOG_PATH = "log.path"
+
+// The log level to use for the logs in htrace.
+const HTRACE_LOG_LEVEL = "log.level"
+
// Default values for HTrace configuration keys.
var DEFAULTS = map[string]string{
HTRACE_WEB_ADDRESS: fmt.Sprintf("0.0.0.0:%d", HTRACE_WEB_ADDRESS_DEFAULT_PORT),
@@ -59,4 +65,6 @@
PATH_LIST_SEP + PATH_SEP + "tmp" + PATH_SEP + "htrace2",
HTRACE_DATA_STORE_CLEAR: "false",
HTRACE_DATA_STORE_SPAN_BUFFER_SIZE: "100",
+ HTRACE_LOG_PATH: "",
+ HTRACE_LOG_LEVEL: "INFO",
}
diff --git a/htrace-core/src/go/src/org/apache/htrace/htraced/datastore.go b/htrace-core/src/go/src/org/apache/htrace/htraced/datastore.go
index 2137063..40678bd 100644
--- a/htrace-core/src/go/src/org/apache/htrace/htraced/datastore.go
+++ b/htrace-core/src/go/src/org/apache/htrace/htraced/datastore.go
@@ -23,7 +23,6 @@
"bytes"
"encoding/gob"
"github.com/jmhodges/levigo"
- "log"
"org/apache/htrace/common"
"org/apache/htrace/conf"
"os"
@@ -163,18 +162,20 @@
// Process incoming spans for a shard.
func (shd *shard) processIncoming() {
+ lg := shd.store.lg
for {
span := <-shd.incoming
if span == nil {
- log.Printf("Shard processor for %s exiting.", shd.path)
+ lg.Infof("Shard processor for %s exiting.\n", shd.path)
shd.exited <- true
return
}
err := shd.writeSpan(span)
if err != nil {
- log.Fatal("Shard processor for %s got fatal error %s.", shd.path, err.Error())
+ lg.Errorf("Shard processor for %s got fatal error %s.\n", shd.path, err.Error())
+ } else {
+ lg.Tracef("Shard processor for %s wrote span %s.\n", shd.path, span.ToJson())
}
- //log.Printf("Shard processor for %s wrote span %s.", shd.path, span.ToJson())
}
}
@@ -236,17 +237,20 @@
// Close a shard.
func (shd *shard) Close() {
+ lg := shd.store.lg
shd.incoming <- nil
- log.Printf("Waiting for %s to exit...", shd.path)
+ lg.Infof("Waiting for %s to exit...\n", shd.path)
if shd.exited != nil {
<-shd.exited
}
shd.ldb.Close()
- log.Printf("Closed %s...", shd.path)
+ lg.Infof("Closed %s...\n", shd.path)
}
// The Data Store.
type dataStore struct {
+ lg *common.Logger
+
// The shards which manage our LevelDB instances.
shards []*shard
@@ -272,7 +276,8 @@
// If we return an error, close the store.
var err error
- store := &dataStore{shards: []*shard{}, WrittenSpans: writtenSpans}
+ lg := common.NewLogger("datastore", cnf)
+ store := &dataStore{lg: lg, shards: []*shard{}, WrittenSpans: writtenSpans}
defer func() {
if err != nil {
store.Close()
@@ -296,21 +301,21 @@
}
if !clearStored {
// TODO: implement re-opening saved data
- log.Println("Error: path " + path + "already exists.")
+ lg.Error("Error: path " + path + "already exists.")
return nil, err
} else {
err = os.RemoveAll(path)
if err != nil {
- log.Println("Failed to create " + path + ": " + err.Error())
+ lg.Error("Failed to create " + path + ": " + err.Error())
return nil, err
}
- log.Println("Cleared " + path)
+ lg.Info("Cleared " + path)
}
}
var shd *shard
shd, err = CreateShard(store, cnf, path)
if err != nil {
- log.Printf("Error creating shard %s: %s", path, err.Error())
+ lg.Errorf("Error creating shard %s: %s", path, err.Error())
return nil, err
}
store.shards = append(store.shards, shd)
@@ -320,7 +325,7 @@
shd := store.shards[idx]
err := shd.WriteMetadata(meta)
if err != nil {
- log.Println("Failed to write metadata to " + store.shards[idx].path + ": " + err.Error())
+ lg.Error("Failed to write metadata to " + store.shards[idx].path + ": " + err.Error())
return nil, err
}
shd.exited = make(chan bool, 1)
@@ -339,7 +344,7 @@
//openOpts.SetFilterPolicy(filter)
ldb, err := levigo.Open(path, openOpts)
if err != nil {
- log.Println("LevelDB failed to open " + path + ": " + err.Error())
+ store.lg.Errorf("LevelDB failed to open %s: %s\n", path, err.Error())
return nil, err
}
defer func() {
@@ -350,7 +355,7 @@
spanBufferSize := cnf.GetInt(conf.HTRACE_DATA_STORE_SPAN_BUFFER_SIZE)
shd = &shard{store: store, ldb: ldb, path: path,
incoming: make(chan *common.Span, spanBufferSize)}
- log.Println("LevelDB opened " + path)
+ store.lg.Infof("LevelDB opened %s\n", path)
return shd, nil
}
@@ -362,12 +367,19 @@
func (store *dataStore) Close() {
for idx := range store.shards {
store.shards[idx].Close()
+ store.shards[idx] = nil
}
if store.readOpts != nil {
store.readOpts.Close()
+ store.readOpts = nil
}
if store.writeOpts != nil {
store.writeOpts.Close()
+ store.writeOpts = nil
+ }
+ if store.lg != nil {
+ store.lg.Close()
+ store.lg = nil
}
}
@@ -385,12 +397,13 @@
}
func (shd *shard) FindSpan(sid int64) *common.Span {
+ lg := shd.store.lg
buf, err := shd.ldb.Get(shd.store.readOpts, makeKey('s', sid))
if err != nil {
if strings.Index(err.Error(), "NotFound:") != -1 {
return nil
}
- log.Printf("Shard(%s): FindSpan(%016x) error: %s\n",
+ lg.Warnf("Shard(%s): FindSpan(%016x) error: %s\n",
shd.path, sid, err.Error())
return nil
}
@@ -399,7 +412,7 @@
data := common.SpanData{}
err = decoder.Decode(&data)
if err != nil {
- log.Printf("Shard(%s): FindSpan(%016x) decode error: %s\n",
+ lg.Errorf("Shard(%s): FindSpan(%016x) decode error: %s\n",
shd.path, sid, err.Error())
return nil
}
@@ -426,7 +439,7 @@
shd := store.shards[idx]
childIds, lim, err = shd.FindChildren(sid, childIds, lim)
if err != nil {
- log.Printf("Shard(%s): FindChildren(%016x) error: %s\n",
+ store.lg.Errorf("Shard(%s): FindChildren(%016x) error: %s\n",
shd.path, sid, err.Error())
}
idx++
diff --git a/htrace-core/src/go/src/org/apache/htrace/htraced/htraced.go b/htrace-core/src/go/src/org/apache/htrace/htraced/htraced.go
index d444a02..4694789 100644
--- a/htrace-core/src/go/src/org/apache/htrace/htraced/htraced.go
+++ b/htrace-core/src/go/src/org/apache/htrace/htraced/htraced.go
@@ -20,8 +20,9 @@
package main
import (
- "log"
+ "org/apache/htrace/common"
"org/apache/htrace/conf"
+ "os"
"time"
)
@@ -30,13 +31,17 @@
func main() {
cnf := conf.LoadApplicationConfig(nil)
+ lg := common.NewLogger("main", cnf)
+ defer lg.Close()
store, err := CreateDataStore(cnf, nil)
if err != nil {
- log.Fatalf("Error creating datastore: %s\n", err.Error())
+ lg.Errorf("Error creating datastore: %s\n", err.Error())
+ os.Exit(1)
}
_, err = CreateRestServer(cnf, store)
if err != nil {
- log.Fatalf("Error creating REST server: %s\n", err.Error())
+ lg.Errorf("Error creating REST server: %s\n", err.Error())
+ os.Exit(1)
}
for {
time.Sleep(time.Duration(10) * time.Hour)
diff --git a/htrace-core/src/go/src/org/apache/htrace/htraced/rest.go b/htrace-core/src/go/src/org/apache/htrace/htraced/rest.go
index d175f4e..efc89e1 100644
--- a/htrace-core/src/go/src/org/apache/htrace/htraced/rest.go
+++ b/htrace-core/src/go/src/org/apache/htrace/htraced/rest.go
@@ -24,7 +24,6 @@
"fmt"
"github.com/gorilla/mux"
"io"
- "log"
"mime"
"net"
"net/http"
@@ -42,23 +41,25 @@
}
// Write a JSON error response.
-func writeError(w http.ResponseWriter, errCode int, errStr string) {
+func writeError(lg *common.Logger, w http.ResponseWriter, errCode int,
+ errStr string) {
str := strings.Replace(errStr, `"`, `'`, -1)
- log.Println(str)
+ lg.Info(str)
w.WriteHeader(errCode)
w.Write([]byte(`{ "error" : "` + str + `"}`))
}
type serverInfoHandler struct {
+ lg *common.Logger
}
-func (handler *serverInfoHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+func (hand *serverInfoHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
setResponseHeaders(w.Header())
version := common.ServerInfo{ReleaseVersion: RELEASE_VERSION,
GitVersion: GIT_VERSION}
buf, err := json.Marshal(&version)
if err != nil {
- writeError(w, http.StatusInternalServerError,
+ writeError(hand.lg, w, http.StatusInternalServerError,
fmt.Sprintf("error marshalling ServerInfo: %s\n", err.Error()))
return
}
@@ -66,13 +67,14 @@
}
type dataStoreHandler struct {
+ lg *common.Logger
store *dataStore
}
func (hand *dataStoreHandler) parse64(w http.ResponseWriter, str string) (int64, bool) {
val, err := strconv.ParseUint(str, 16, 64)
if err != nil {
- writeError(w, http.StatusBadRequest,
+ writeError(hand.lg, w, http.StatusBadRequest,
fmt.Sprintf("Failed to parse span ID %s: %s", str, err.Error()))
w.Write([]byte("Error parsing : " + err.Error()))
return -1, false
@@ -84,12 +86,12 @@
req *http.Request) (int32, bool) {
str := req.FormValue(fieldName)
if str == "" {
- writeError(w, http.StatusBadRequest, fmt.Sprintf("No %s specified.", fieldName))
+ writeError(hand.lg, w, http.StatusBadRequest, fmt.Sprintf("No %s specified.", fieldName))
return -1, false
}
val, err := strconv.ParseUint(str, 16, 32)
if err != nil {
- writeError(w, http.StatusBadRequest,
+ writeError(hand.lg, w, http.StatusBadRequest,
fmt.Sprintf("Error parsing %s: %s.", fieldName, err.Error()))
return -1, false
}
@@ -111,7 +113,7 @@
}
span := hand.store.FindSpan(sid)
if span == nil {
- writeError(w, http.StatusNoContent, fmt.Sprintf("No such span as %s",
+ writeError(hand.lg, w, http.StatusNoContent, fmt.Sprintf("No such span as %s\n",
common.SpanId(sid)))
return
}
@@ -157,7 +159,7 @@
err := dec.Decode(&span)
if err != nil {
if err != io.EOF {
- writeError(w, http.StatusBadRequest,
+ writeError(hand.lg, w, http.StatusBadRequest,
fmt.Sprintf("Error parsing spans: %s", err.Error()))
return
}
@@ -166,12 +168,13 @@
spans = append(spans, &span)
}
for spanIdx := range spans {
- log.Printf("writing span %s\n", spans[spanIdx].ToJson())
+ hand.lg.Debugf("writing span %s\n", spans[spanIdx].ToJson())
hand.store.WriteSpan(spans[spanIdx])
}
}
type defaultServeHandler struct {
+ lg *common.Logger
}
func (hand *defaultServeHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
@@ -182,7 +185,7 @@
ident = strings.Replace(ident, "/", "__", -1)
rsc := resource.Catalog[ident]
if rsc == "" {
- log.Printf("failed to find entry for %s\n", ident)
+ hand.lg.Warnf("failed to find entry for %s\n", ident)
w.WriteHeader(http.StatusNotFound)
return
}
@@ -194,6 +197,7 @@
type RestServer struct {
listener net.Listener
+ lg *common.Logger
}
func CreateRestServer(cnf *conf.Config, store *dataStore) (*RestServer, error) {
@@ -203,26 +207,36 @@
if err != nil {
return nil, err
}
+ var success bool
+ defer func() {
+ if !success {
+ rsv.Close()
+ }
+ }()
+ rsv.lg = common.NewLogger("rest", cnf)
r := mux.NewRouter().StrictSlash(false)
// Default Handler. This will serve requests for static requests.
- r.Handle("/", &defaultServeHandler{})
+ r.Handle("/", &defaultServeHandler{lg: rsv.lg})
- r.Handle("/server/info", &serverInfoHandler{}).Methods("GET")
+ r.Handle("/server/info", &serverInfoHandler{lg: rsv.lg}).Methods("GET")
- writeSpansH := &writeSpansHandler{dataStoreHandler: dataStoreHandler{store: store}}
+ writeSpansH := &writeSpansHandler{dataStoreHandler: dataStoreHandler{
+ store: store, lg: rsv.lg}}
r.Handle("/writeSpans", writeSpansH).Methods("POST")
span := r.PathPrefix("/span").Subrouter()
- findSidH := &findSidHandler{dataStoreHandler: dataStoreHandler{store: store}}
+ findSidH := &findSidHandler{dataStoreHandler: dataStoreHandler{store: store, lg: rsv.lg}}
span.Handle("/{id}", findSidH).Methods("GET")
- findChildrenH := &findChildrenHandler{dataStoreHandler: dataStoreHandler{store: store}}
+ findChildrenH := &findChildrenHandler{dataStoreHandler: dataStoreHandler{store: store,
+ lg: rsv.lg}}
span.Handle("/{id}/children", findChildrenH).Methods("GET")
go http.Serve(rsv.listener, r)
- log.Println("Started REST server...")
+ rsv.lg.Infof("Started REST server on %s...\n", rsv.listener.Addr().String())
+ success = true
return rsv, nil
}
