GET /v1/o11y/logs reads event.log's columns for the caller's org

GetRecentLogs selected timestamp, resources_string and resource, which
event.log does not have, so every call answered 500 "Unknown expression
identifier timestamp". It now selects time, service and the k8s labels
from the attributes map, and binds org, the window and the limit as
arguments. The route answers 403 when the request names no tenant.

Co-authored-by: Hanzo Dev <[email protected]>

pkg/query-service/app/datastorereader/reader.go

@@ -23,6 +23,7 @@ import (
 	"github.com/hanzoai/o11y/pkg/sqlstore"
 	"github.com/hanzoai/o11y/pkg/telemetrystore"
 	"github.com/hanzoai/o11y/pkg/types"
+	"github.com/hanzoai/o11y/pkg/types/authtypes"
 	"github.com/hanzoai/o11y/pkg/types/ctxtypes"
 	"github.com/hanzoai/o11y/pkg/types/featuretypes"
 	"github.com/hanzoai/o11y/pkg/types/instrumentationtypes"
@@ -4184,11 +4185,16 @@ func (r *DatastoreReader) GetTimeSeriesResultV3(ctx context.Context, query strin
 
 // GetListResultV3 runs the query and returns list of rows
 func (r *DatastoreReader) GetListResultV3(ctx context.Context, query string) ([]*v3.Row, error) {
+	return r.listRows(ctx, query)
+}
+
+// listRows runs query with its bound args and scans each row into a v3.Row.
+func (r *DatastoreReader) listRows(ctx context.Context, query string, args ...any) ([]*v3.Row, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
 		instrumentationtypes.CodeNamespace:    "datastore-reader",
 		instrumentationtypes.CodeFunctionName: "GetListResultV3",
 	})
-	rows, err := r.db.Query(ctx, query)
+	rows, err := r.db.Query(ctx, query, args...)
 	if err != nil {
 		r.logger.Error("error while reading time series result", errorsV2.Attr(err))
 		return nil, errors.New(err.Error())
@@ -5600,15 +5606,17 @@ func (r *DatastoreReader) GetNormalizedStatus(
 	return result, nil
 }
 
-// GetRecentLogs reads the most recent logs in [startNano, endNano], newest first,
-// capped at limit, from the configured logs table (r.logsDB.r.logsTableV2) — the
-// real read behind GET /v1/o11y/logs, replacing the empty
-// {"results":[]} stub. The bounds are int64 nanosecond epochs and limit is an int
-// (never user strings), so the fmt.Sprintf interpolation is injection-safe. It
-// selects the core log columns plus service.name and the k8s namespace/pod/
-// container the OTel collector tags each line with, and reuses the shared
-// GetListResultV3 row scanner (one row-reading path).
+// GetRecentLogs reads the tenant's most recent logs in [startNano, endNano],
+// newest first, capped at limit, from the configured logs table — the read
+// behind GET /v1/o11y/logs. The tenant comes from ctx and is bound as
+// `org = ?`; a ctx without one is refused. The columns are event.log's own:
+// time, service, and the attributes map that carries the k8s resource labels.
+// Rows are scanned by the shared v3 row reader.
 func (r *DatastoreReader) GetRecentLogs(ctx context.Context, startNano, endNano int64, limit int) ([]*v3.Row, error) {
+	tenant, err := authtypes.TenantFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
 	if limit <= 0 {
 		limit = 100
 	}
@@ -5621,16 +5629,18 @@ func (r *DatastoreReader) GetRecentLogs(ctx context.Context, startNano, endNano
 	if startNano <= 0 || startNano >= endNano {
 		startNano = endNano - int64(15*time.Minute)
 	}
-	table := r.logsDB + "." + r.logsTableV2
-	query := fmt.Sprintf(
-		"SELECT timestamp, id, trace_id, span_id, severity_text, severity_number, body, "+
-			"resources_string['service.name'] AS service, "+
-			"resource.`service.name`::String AS service_name, "+
-			"resource.`k8s.namespace.name`::String AS k8s_namespace, "+
-			"resource.`k8s.pod.name`::String AS k8s_pod, "+
-			"resource.`k8s.container.name`::String AS k8s_container "+
-			"FROM %s WHERE timestamp >= %d AND timestamp <= %d "+
-			"ORDER BY timestamp DESC LIMIT %d",
-		table, startNano, endNano, limit)
-	return r.GetListResultV3(ctx, query)
+	return r.listRows(ctx, recentLogsQuery(r.logsDB+"."+r.logsTableV2),
+		tenant, fmt.Sprintf("%d", startNano), fmt.Sprintf("%d", endNano), limit)
+}
+
+// recentLogsQuery is GetRecentLogs' statement over table. time is DateTime64(9),
+// so the nanosecond bounds compare against it directly.
+func recentLogsQuery(table string) string {
+	return "SELECT time AS timestamp, id, trace_id, span_id, severity_text, severity_number, body, " +
+		"service, service AS service_name, " +
+		"attributes['k8s.namespace.name'] AS k8s_namespace, " +
+		"attributes['k8s.pod.name'] AS k8s_pod, " +
+		"attributes['k8s.container.name'] AS k8s_container " +
+		"FROM " + table + " WHERE org = ? AND time >= ? AND time <= ? " +
+		"ORDER BY time DESC LIMIT ?"
 }

pkg/query-service/app/datastorereader/recent_logs_test.go

@@ -0,0 +1,58 @@
+package datastorereader
+
+import (
+	"context"
+	"regexp"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	dsmock "github.com/hanzo-ds/mock"
+	"github.com/hanzoai/o11y/pkg/errors"
+	"github.com/hanzoai/o11y/pkg/instrumentation/instrumentationtest"
+	"github.com/hanzoai/o11y/pkg/telemetrystore"
+	"github.com/hanzoai/o11y/pkg/telemetrystore/telemetrystoretest"
+	"github.com/hanzoai/o11y/pkg/types/authtypes"
+	"github.com/stretchr/testify/require"
+)
+
+// GetRecentLogs reads event.log's own columns for the tenant on ctx, binding
+// the tenant, the window and the limit.
+func TestGetRecentLogsReadsTheTenantsRows(t *testing.T) {
+	store := telemetrystoretest.New(telemetrystore.Config{}, sqlmock.QueryMatcherEqual)
+	r := &DatastoreReader{db: store.Datastore(), logsDB: "event", logsTableV2: "log", logger: instrumentationtest.New().Logger()}
+
+	at := time.Unix(1790290369, 0).UTC()
+	store.Mock().ExpectQuery(recentLogsQuery("event.log")).
+		WithArgs("acme", "1790290000000000000", "1790291000000000000", 5).
+		WillReturnRows(dsmock.NewRows([]dsmock.ColumnType{
+			{Name: "timestamp", Type: "DateTime"},
+			{Name: "id", Type: "String"},
+			{Name: "body", Type: "String"},
+			{Name: "service", Type: "String"},
+		}, [][]any{{at, "row-1", "request", "cloud"}}))
+
+	ctx := authtypes.NewContextWithTenant(context.Background(), "acme")
+	rows, err := r.GetRecentLogs(ctx, 1790290000000000000, 1790291000000000000, 5)
+	require.NoError(t, err)
+	require.Len(t, rows, 1)
+	require.Equal(t, at, rows[0].Timestamp)
+	require.NoError(t, store.Mock().ExpectationsWereMet())
+}
+
+// The statement names the envelope columns, never the dropped ones, and
+// scopes by org.
+func TestRecentLogsQueryUsesTheEnvelopeColumns(t *testing.T) {
+	q := recentLogsQuery("event.log")
+	require.Regexp(t, regexp.MustCompile(`^SELECT time AS timestamp, .* FROM event\.log WHERE org = \? AND time >= \? AND time <= \? ORDER BY time DESC LIMIT \?$`), q)
+	for _, dropped := range []string{"resources_string", "resource.`", "timestamp >="} {
+		require.NotContains(t, q, dropped)
+	}
+}
+
+// A ctx with no tenant is refused before any read.
+func TestGetRecentLogsRefusesWithoutATenant(t *testing.T) {
+	r := &DatastoreReader{logsDB: "event", logsTableV2: "log"}
+	_, err := r.GetRecentLogs(context.Background(), 0, 0, 10)
+	require.True(t, errors.Ast(err, errors.TypeForbidden), "got %v", err)
+}

pkg/query-service/app/http_handler.go

@@ -2579,8 +2579,9 @@ func (aH *APIHandler) logFieldUpdate(w http.ResponseWriter, r *http.Request) {
 // query window, newest first. Reads the configured logs table via the reader
 // (real Datastore rows), replacing the former {"results":[]} stub. Params:
 // limit (default 100, max 1000), timestampStart / timestampEnd (nanosecond
-// epochs; default the last 15 minutes). Honest: a read error surfaces as an
-// error, never a fabricated empty list.
+// epochs; default the last 15 minutes). The read is the caller's tenant only;
+// a request that carries none is 403. A read error surfaces as an error, never
+// a fabricated empty list.
 func (aH *APIHandler) getLogs(w http.ResponseWriter, r *http.Request) {
 	q := r.URL.Query()
 	limit := 100
@@ -2596,6 +2597,10 @@ func (aH *APIHandler) getLogs(w http.ResponseWriter, r *http.Request) {
 		start = v
 	}
 
+	if _, err := authtypes.TenantFromContext(r.Context()); err != nil {
+		RespondError(w, &model.ApiError{Typ: model.ErrorForbidden, Err: err}, nil)
+		return
+	}
 	rows, err := aH.reader.GetRecentLogs(r.Context(), start, end, limit)
 	if err != nil {
 		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, "Failed to fetch logs from the DB")