server+runtime+logs: Add the req_id attribute on the decision logs (#…

…5196)

Today it is not possible to correlate the decision log with
other types of logs (server, print, etc.) when the server log
level is >= INFO. The log correlation could be helpful in
troubleshooting.

A solution is to add a common attribute in all logs to make
the log correlation possible, so adding the req_id attribute
on decision logs, when server log level is >= INFO, will make it
possible.

Fixes: #5006

* Add documentation about decision log req_id attribute

The documentation purpose is to explain the relation with
others logs, how it could be used, and when it is included
on decision logs.

Signed-off-by: Humberto Corrêa da Silva <humbertoc_silva@hotmail.com>

docs/content/management-decision-logs.md

@@ -76,6 +76,7 @@ Decision log updates contain the following fields:
 | `[_].erased` | `array[string]` | Set of JSON Pointers specifying fields in the event that were erased. |
 | `[_].masked` | `array[string]` | Set of JSON Pointers specifying fields in the event that were masked. |
 | `[_].nd_builtin_cache` | `object` | Key-value pairs of non-deterministic builtin names, paired with objects specifying the input/output mappings for each unique invocation of that builtin during policy evaluation. Intended for use in debugging and decision replay. Receivers will need to decode the JSON using Rego's JSON decoders. |
+| `[_].req_id` | `number` | Incremental request identifier, and unique only to the OPA instance, for the request that started the policy query. The attribute value is the same as the value present in others logs (request, response, and print) and could be used to correlate them all. This attribute will be included just when OPA runtime is initialized in server mode and the log level is equal to or greater than info. |
 
 If the decision log was successfully uploaded to the remote service, it should respond with an HTTP 2xx status. If the
 service responds with a non-2xx status, OPA will requeue the last chunk containing decision log events and upload it

logging/logging.go

@@ -1,6 +1,7 @@
 package logging
 
 import (
+	"context"
 	"io"
 
 	"github.com/sirupsen/logrus"
@@ -185,3 +186,37 @@ func (l *NoOpLogger) SetLevel(level Level) {
 func (l *NoOpLogger) GetLevel() Level {
 	return l.level
 }
+
+type requestContextKey string
+
+const reqCtxKey = requestContextKey("request-context-key")
+
+// RequestContext represents the request context used to store data
+// related to the request that could be used on logs.
+type RequestContext struct {
+	ClientAddr string
+	ReqID      uint64
+	ReqMethod  string
+	ReqPath    string
+}
+
+// Fields adapts the RequestContext fields to logrus.Fields.
+func (rctx RequestContext) Fields() logrus.Fields {
+	return logrus.Fields{
+		"client_addr": rctx.ClientAddr,
+		"req_id":      rctx.ReqID,
+		"req_method":  rctx.ReqMethod,
+		"req_path":    rctx.ReqPath,
+	}
+}
+
+// NewContext returns a copy of parent with an associated RequestContext.
+func NewContext(parent context.Context, val *RequestContext) context.Context {
+	return context.WithValue(parent, reqCtxKey, val)
+}
+
+// FromContext returns the RequestContext associated with ctx, if any.
+func FromContext(ctx context.Context) (*RequestContext, bool) {
+	requestContext, ok := ctx.Value(reqCtxKey).(*RequestContext)
+	return requestContext, ok
+}

logging/logging_test.go

@@ -82,3 +82,47 @@ func TestWithFieldsMerges(t *testing.T) {
 		t.Fatal("Logger did not contain configured field value")
 	}
 }
+
+func TestRequestContextFields(t *testing.T) {
+	fields := RequestContext{
+		ClientAddr: "127.0.0.1",
+		ReqID:      1,
+		ReqMethod:  "GET",
+		ReqPath:    "/test",
+	}.Fields()
+
+	var fieldvalue interface{}
+	var ok bool
+
+	if fieldvalue, ok = fields["client_addr"]; !ok {
+		t.Fatal("Fields did not contain the client_addr field")
+	}
+
+	if fieldvalue.(string) != "127.0.0.1" {
+		t.Fatal("Fields did not contain the configured client_addr value")
+	}
+
+	if fieldvalue, ok = fields["req_id"]; !ok {
+		t.Fatal("Fields did not contain the req_id field")
+	}
+
+	if fieldvalue.(uint64) != 1 {
+		t.Fatal("Fields did not contain the configured req_id value")
+	}
+
+	if fieldvalue, ok = fields["req_method"]; !ok {
+		t.Fatal("Fields did not contain the req_method field")
+	}
+
+	if fieldvalue.(string) != "GET" {
+		t.Fatal("Fields did not contain the configured req_method value")
+	}
+
+	if fieldvalue, ok = fields["req_path"]; !ok {
+		t.Fatal("Fields did not contain the req_path field")
+	}
+
+	if fieldvalue.(string) != "/test" {
+		t.Fatal("Fields did not contain the configured req_path value")
+	}
+}

plugins/logs/plugin.go

@@ -61,6 +61,7 @@ type EventV1 struct {
 	RequestedBy    string                  `json:"requested_by,omitempty"`
 	Timestamp      time.Time               `json:"timestamp"`
 	Metrics        map[string]interface{}  `json:"metrics,omitempty"`
+	RequestID      uint64                  `json:"req_id,omitempty"`
 
 	inputAST ast.Value
 }
@@ -96,6 +97,7 @@ var errorKey = ast.StringTerm("error")
 var requestedByKey = ast.StringTerm("requested_by")
 var timestampKey = ast.StringTerm("timestamp")
 var metricsKey = ast.StringTerm("metrics")
+var requestIDKey = ast.StringTerm("req_id")
 
 // AST returns the Rego AST representation for a given EventV1 object.
 // This avoids having to round trip through JSON while applying a decision log
@@ -214,6 +216,10 @@ func (e *EventV1) AST() (ast.Value, error) {
 		event.Insert(metricsKey, ast.NewTerm(m))
 	}
 
+	if e.RequestID > 0 {
+		event.Insert(requestIDKey, ast.UIntNumberTerm(e.RequestID))
+	}
+
 	return event, nil
 }
 
@@ -597,6 +603,7 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) error {
 		NDBuiltinCache: decision.NDBuiltinCache,
 		RequestedBy:    decision.RemoteAddr,
 		Timestamp:      decision.Timestamp,
+		RequestID:      decision.RequestID,
 		inputAST:       decision.InputAST,
 	}
 

plugins/logs/plugin_test.go

@@ -2266,6 +2266,25 @@ func TestEventV1ToAST(t *testing.T) {
 				NDBuiltinCache: &ndbCacheExample,
 			},
 		},
+		{
+			note: "event with req id",
+			event: EventV1{
+				Labels:     map[string]string{"foo": "1", "bar": "2"},
+				DecisionID: "1234567890",
+				Bundles: map[string]BundleInfoV1{
+					"b1": {"revision7"},
+					"b2": {"0"},
+					"b3": {},
+				},
+				Input:       &goInput,
+				Path:        "/http/authz/allow",
+				RequestedBy: "[::1]:59943",
+				Result:      &result,
+				Timestamp:   time.Now(),
+				RequestID:   1,
+				inputAST:    astInput,
+			},
+		},
 	}
 
 	for _, tc := range cases {

runtime/logging.go

@@ -6,7 +6,6 @@ package runtime
 
 import (
 	"bytes"
-	"context"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -16,7 +15,6 @@ import (
 
 	"github.com/open-policy-agent/opa/logging"
 	"github.com/open-policy-agent/opa/topdown/print"
-	"github.com/sirupsen/logrus"
 )
 
 type loggingPrintHook struct {
@@ -26,7 +24,7 @@ type loggingPrintHook struct {
 func (h loggingPrintHook) Print(pctx print.Context, msg string) error {
 	// NOTE(tsandall): if the request context is not present then do not panic,
 	// just log the print message without the additional context.
-	rctx, _ := pctx.Context.Value(reqCtxKey).(requestContext)
+	rctx, _ := logging.FromContext(pctx.Context)
 	fields := rctx.Fields()
 	fields["line"] = pctx.Location.String()
 	h.logger.WithFields(fields).Info(msg)
@@ -54,28 +52,8 @@ func (h *LoggingHandler) loggingEnabled(level logging.Level) bool {
 	return level <= h.logger.GetLevel()
 }
 
-type requestContextKey string
-
-const reqCtxKey = requestContextKey("request-context-key")
-
-type requestContext struct {
-	ClientAddr string
-	ReqID      uint64
-	ReqMethod  string
-	ReqPath    string
-}
-
-func (rctx requestContext) Fields() logrus.Fields {
-	return logrus.Fields{
-		"client_addr": rctx.ClientAddr,
-		"req_id":      rctx.ReqID,
-		"req_method":  rctx.ReqMethod,
-		"req_path":    rctx.ReqPath,
-	}
-}
-
 func (h *LoggingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	var rctx requestContext
+	var rctx logging.RequestContext
 	rctx.ReqID = atomic.AddUint64(&h.requestID, uint64(1))
 	recorder := newRecorder(h.logger, w, r, rctx.ReqID, h.loggingEnabled(logging.Debug))
 	t0 := time.Now()
@@ -85,7 +63,7 @@ func (h *LoggingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		rctx.ClientAddr = r.RemoteAddr
 		rctx.ReqMethod = r.Method
 		rctx.ReqPath = r.URL.EscapedPath()
-		r = r.WithContext(context.WithValue(r.Context(), reqCtxKey, rctx))
+		r = r.WithContext(logging.NewContext(r.Context(), &rctx))
 
 		var err error
 		fields := rctx.Fields()

server/buffer.go

@@ -31,6 +31,7 @@ type Info struct {
 	Error          error
 	Metrics        metrics.Metrics
 	Trace          []*topdown.Event
+	RequestID      uint64
 }
 
 // BundleInfo contains information describing a bundle.

server/server.go

@@ -35,6 +35,7 @@ import (
 	"github.com/open-policy-agent/opa/ast"
 	"github.com/open-policy-agent/opa/bundle"
 	"github.com/open-policy-agent/opa/internal/json/patch"
+	"github.com/open-policy-agent/opa/logging"
 	"github.com/open-policy-agent/opa/metrics"
 	"github.com/open-policy-agent/opa/plugins"
 	bundlePlugin "github.com/open-policy-agent/opa/plugins/bundle"
@@ -2908,6 +2909,11 @@ func (l decisionLogger) Log(ctx context.Context, txn storage.Transaction, decisi
 		bundles[name] = BundleInfo{Revision: rev}
 	}
 
+	var reqID uint64
+	if rctx, ok := logging.FromContext(ctx); ok {
+		reqID = rctx.ReqID
+	}
+
 	info := &Info{
 		Txn:        txn,
 		Revision:   l.revision,
@@ -2922,6 +2928,7 @@ func (l decisionLogger) Log(ctx context.Context, txn storage.Transaction, decisi
 		Results:    goResults,
 		Error:      err,
 		Metrics:    m,
+		RequestID:  reqID,
 	}
 
 	if ndbCache != nil {