LogDog: Add project namespace to logs endpoint.

appengine/logdog/coordinator/endpoints/logs/get.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package logs
 
 import (
-        "net/url"
         "time"
 
         "github.com/golang/protobuf/proto"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/common/api/logdog_coordinator/logs/v1"
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/common/retry"
         "github.com/luci/luci-go/server/logdog/storage"
         "github.com/luci/luci-go/server/logdog/storage/archive"
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 )
@@ skipping 14 matching lines @@
 )
 
 // Get returns state and log data for a single log stream.
 func (s *server) Get(c context.Context, req *logdog.GetRequest) (*logdog.GetResponse, error) {
         return s.getImpl(c, req, false)
 }
 
 // Tail returns the last log entry for a given log stream.
 func (s *server) Tail(c context.Context, req *logdog.TailRequest) (*logdog.GetResponse, error) {
         r := logdog.GetRequest{
-                Path:  req.Path,
-                State: req.State,
+                Project: req.Project,
+                Path:    req.Path,
+                State:   req.State,
         }
         return s.getImpl(c, &r, true)
 }
 
 // getImpl is common code shared between Get and Tail endpoints.
 func (s *server) getImpl(c context.Context, req *logdog.GetRequest, tail bool) (*logdog.GetResponse, error) {
         log.Fields{
-                "path":  req.Path,
-                "index": req.Index,
-                "tail":  tail,
+                "project": req.Project,
+                "path":    req.Path,
+                "index":   req.Index,
+                "tail":    tail,
         }.Debugf(c, "Received get request.")
 
-        // Fetch the log stream state for this log stream.
-        u, err := url.Parse(req.Path)
+        ls, err := coordinator.NewLogStream(req.Path)
         if err != nil {
-                log.Fields{
-                        log.ErrorKey: err,
-                        "path":       req.Path,
-                }.Errorf(c, "Could not parse path URL.")
-                return nil, grpcutil.Errf(codes.InvalidArgument, "invalid path encoding")
-        }
-        ls, err := coordinator.NewLogStream(u.Path)
-        if err != nil {
-                log.Fields{
-                        log.ErrorKey: err,
-                        "path":       u.Path,
-                }.Errorf(c, "Invalid path supplied.")
+                log.WithError(err).Errorf(c, "Invalid path supplied.")
                 return nil, grpcutil.Errf(codes.InvalidArgument, "invalid path value")
         }
 
+        // The user may supply a hash instead of a full path. Once resolved, log
+        // the original log stream.
+        path := ls.Path()
+        if req.Path != string(path) {
+                log.Fields{
+                        "hashPath":   req.Path,
+                        "streamPath": path,
+                }.Debugf(c, "Resolved hash path.")
+        }
+
         // If this log entry is Purged and we're not admin, pretend it doesn't exist.
         err = ds.Get(c).Get(ls)
         switch err {
         case nil:
                 if ls.Purged {
                         if authErr := coordinator.IsAdminUser(c); authErr != nil {
                                 log.Fields{
                                         log.ErrorKey: authErr,
                                 }.Warningf(c, "Non-superuser requested purged log.")
                                 return nil, grpcutil.Errf(codes.NotFound, "path not found")
                         }
                 }
 
         case ds.ErrNoSuchEntity:
-                log.Fields{
-                        "path": u.Path,
-                }.Errorf(c, "Log stream does not exist.")
+                log.Errorf(c, "Log stream does not exist.")
                 return nil, grpcutil.Errf(codes.NotFound, "path not found")
 
         default:
-                log.Fields{
-                        log.ErrorKey: err,
-                        "path":       u.Path,
-                }.Errorf(c, "Failed to look up log stream.")
+                log.WithError(err).Errorf(c, "Failed to look up log stream.")
                 return nil, grpcutil.Internal
         }
-        path := ls.Path()
 
         // If nothing was requested, return nothing.
         resp := logdog.GetResponse{}
         if !(req.State || tail) && req.LogCount < 0 {
                 return &resp, nil
         }
 
         if req.State {
                 resp.State = loadLogStreamState(ls)
 
                 var err error
                 resp.Desc, err = ls.DescriptorValue()
                 if err != nil {
                         log.WithError(err).Errorf(c, "Failed to deserialize descriptor protobuf.")
                         return nil, grpcutil.Internal
                 }
         }
 
         // Retrieve requested logs from storage, if requested.
         if tail || req.LogCount >= 0 {
                 resp.Logs, err = s.getLogs(c, req, tail, ls)
                 if err != nil {
-                        log.Fields{
-                                log.ErrorKey: err,
-                                "path":       path,
-                        }.Errorf(c, "Failed to get logs.")
+                        log.WithError(err).Errorf(c, "Failed to get logs.")
                         return nil, grpcutil.Internal
                 }
         }
 
         log.Fields{
                 "logCount": len(resp.Logs),
         }.Debugf(c, "Get request completed successfully.")
         return &resp, nil
 }
 
@@ skipping 40 matching lines @@
                         Client:    gs,
                         MaxBytes:  byteLimit,
                 })
                 if err != nil {
                         log.WithError(err).Errorf(c, "Failed to create Google Storage storage instance.")
                         return nil, err
                 }
         }
         defer st.Close()
 
-        path := ls.Path()
+        project, path := coordinator.Project(c), ls.Path()
 
         var fetchedLogs [][]byte
         var err error
         if tail {
-                fetchedLogs, err = getTail(c, st, path)
+                fetchedLogs, err = getTail(c, st, project, path)
         } else {
-                fetchedLogs, err = getHead(c, req, st, path, byteLimit)
+                fetchedLogs, err = getHead(c, req, st, project, path, byteLimit)
         }
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to fetch log records.")
                 return nil, err
         }
 
         logEntries := make([]*logpb.LogEntry, len(fetchedLogs))
         for idx, ld := range fetchedLogs {
                 // Deserialize the log entry, then convert it to output value.
                 le := logpb.LogEntry{}
                 if err := proto.Unmarshal(ld, &le); err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
                                 "index":      idx,
                         }.Errorf(c, "Failed to generate response log entry.")
                         return nil, err
                 }
                 logEntries[idx] = &le
         }
         return logEntries, nil
 }
 
-func getHead(c context.Context, req *logdog.GetRequest, st storage.Storage, p types.StreamPath, byteLimit int) (
-        [][]byte, error) {
-        c = log.SetFields(c, log.Fields{
-                "path":          p,
+func getHead(c context.Context, req *logdog.GetRequest, st storage.Storage, project config.ProjectName,
+        path types.StreamPath, byteLimit int) ([][]byte, error) {
+        log.Fields{
+                "project":       project,
+                "path":          path,
                 "index":         req.Index,
                 "count":         req.LogCount,
                 "bytes":         req.ByteCount,
                 "noncontiguous": req.NonContiguous,
-        })
+        }.Debugf(c, "Issuing Get request.")
 
         // Allocate result logs array.
         logCount := int(req.LogCount)
         asz := getInitialArraySize
         if logCount > 0 && logCount < asz {
                 asz = logCount
         }
         logs := make([][]byte, 0, asz)
 
         sreq := storage.GetRequest{
-                Path:  p,
-                Index: types.MessageIndex(req.Index),
-                Limit: logCount,
+                Project: project,
+                Path:    path,
+                Index:   types.MessageIndex(req.Index),
+                Limit:   logCount,
         }
 
         count := 0
         err := retry.Retry(c, retry.TransientOnly(retry.Default), func() error {
                 // Issue the Get request. This may return a transient error, in which case
                 // we will retry.
                 return st.Get(sreq, func(idx types.MessageIndex, ld []byte) bool {
                         if count > 0 && byteLimit-len(ld) < 0 {
                                 // Not the first log, and we've exceeded our byte limit.
                                 return false
@@ skipping 10 matching lines @@
                 })
         }, func(err error, delay time.Duration) {
                 log.Fields{
                         log.ErrorKey:   err,
                         "delay":        delay,
                         "initialIndex": req.Index,
                         "nextIndex":    sreq.Index,
                         "count":        len(logs),
                 }.Warningf(c, "Transient error while loading logs; retrying.")
         })
-        if err != nil {
+        switch err {
+        case nil:
+                return logs, nil
+
+        case storage.ErrDoesNotExist:
+                return nil, nil
+
+        default:
                 log.Fields{
                         log.ErrorKey:   err,
                         "initialIndex": req.Index,
                         "nextIndex":    sreq.Index,
                         "count":        len(logs),
                 }.Errorf(c, "Failed to execute range request.")
                 return nil, err
         }
-
-        return logs, nil
 }
 
-func getTail(c context.Context, st storage.Storage, p types.StreamPath) ([][]byte, error) {
+func getTail(c context.Context, st storage.Storage, project config.ProjectName, path types.StreamPath) ([][]byte, error) {
+        log.Fields{
+                "project": project,
+                "path":    path,
+        }.Debugf(c, "Issuing Tail request.")
+
         var data []byte
         err := retry.Retry(c, retry.TransientOnly(retry.Default), func() (err error) {
-                data, _, err = st.Tail("", p)
+                data, _, err = st.Tail(project, path)
                 return
         }, func(err error, delay time.Duration) {
                 log.Fields{
                         log.ErrorKey: err,
                         "delay":      delay,
                 }.Warningf(c, "Transient error while fetching tail log; retrying.")
         })
-        if err != nil {
+        switch err {
+        case nil:
+                return [][]byte{data}, nil
+
+        case storage.ErrDoesNotExist:
+                return nil, nil
+
+        default:
                 log.WithError(err).Errorf(c, "Failed to fetch tail log.")
                 return nil, err
         }
-        return [][]byte{data}, err
 }