Add LogDog log stream fetcher code.

web/inc/logdog-stream-view/fetcher.ts

+/*
+  Copyright 2016 The LUCI Authors. All rights reserved.
+  Use of this source code is governed under the Apache License, Version 2.0
+  that can be found in the LICENSE file.
+*/
+
+///<reference path="../logdog-stream/logdog.ts" />
+///<reference path="../luci-sleep-promise/promise.ts" />
+///<reference path="../rpc/client.ts" />
+
+namespace LogDog {
+
+  export type FetcherOptions = {
+    byteCount?: number; logCount?: number; sparse?: boolean;
+  };
+
+  // Type of a "Get" or "Tail" response (protobuf).
+  type GetResponse = {state: any; desc: any; logs: any[];};
+
+  /** The Fetcher's current status. */
+  export enum FetcherStatus {
+    // Not doing anything.
+    IDLE,
+    // Attempting to load log data.
+    LOADING,
+    // We're waiting for the log stream to emit more logs.
+    STREAMING,
+    // The log stream is missing.
+    MISSING,
+    // The log stream encountered an error.
+    ERROR
+  }
+
+  /**
+   * Fetcher is responsible for fetching LogDog log stream entries from the
+   * remote service via an RPC client.
+   *
+   * Fetcher is responsible for wrapping the raw RPC calls and their results,
+   * and retrying calls due to: - Transient failures (via RPC client). - Missing

[hinoka, 2017/02/27 21:50:25]

newline for bullet points.

[dnj, 2017/03/08 04:13:43]

Done.

+   * stream (assumption is that the stream is still being ingested and
+   *   registered, and therefore a repeated retry is appropriate).
+   * - Streaming stream (log stream is not terminated, but more records are not
+   *   yet available).
+   *
+   * The interface that Fetcher presents to its caller is a simple Promise-based
+   * method to retrieve log stream data.
+   *
+   * Fetcher offers fetching via "get", "getAll", and "tail".

[hinoka, 2017/02/27 21:50:25]

how about "getLast" instead of "tail"?

[dnj, 2017/03/08 04:13:43]

It's not the last though. Maybe "latest"?

+   */
+  export class Fetcher {
+    private debug = false;
+    private static maxLogsPerGet = 0;
+
+    private lastDesc: LogDog.LogStreamDescriptor;
+    private lastState: LogDog.LogStreamState;
+
+    private currentStatus: FetcherStatus = FetcherStatus.IDLE;
+    private lastErrorValue: Error|null;
+    private statusChangedCallback: (() => void);
+
+    private static missingRetry: luci.Retry = {delay: 5000, maxDelay: 15000};
+    private static streamingRetry: luci.Retry = {delay: 1000, maxDelay: 5000};
+
+    constructor(private client: luci.Client, readonly stream: LogDog.Stream) {}
+
+    get desc() {
+      return this.lastDesc;
+    }
+    get state() {
+      return this.lastState;
+    }
+
+    get status() {
+      return this.currentStatus;
+    }
+
+    /**
+     * Returns the log stream's terminal index.
+     *
+     * If no terminal index is known, or if the log stream is still streaming,

[hinoka, 2017/02/27 21:50:26]

"If no terminal index is known because the log is still streaming"?

[dnj, 2017/03/08 04:13:43]

Done.

+     * this will return -1.
+     */
+    get terminalIndex(): number {
+      return ((this.lastState) ? this.lastState.terminalIndex : -1);
+    }
+
+    /** Archived returns true if this log stream is known to be archived. */
+    get archived(): boolean {
+      return (!!(this.lastState && this.lastState.archive));

[hinoka, 2017/02/27 21:50:25]

does bool(...) exist?  That would be more readable

[dnj, 2017/03/08 04:13:43]

No :(

+    }
+
+    get lastError(): Error|null {
+      return this.lastErrorValue;
+    }
+
+    private setCurrentStatus(st: FetcherStatus, err?: Error) {
+      if (st !== this.currentStatus || err !== this.lastErrorValue) {
+        this.currentStatus = st;

[hinoka, 2017/02/27 21:50:25]

If these are set together, should they be called either "currentStatus" and
"currentError" or "lastStatus" and "lastError", otherwise the naming implies
they're set one-off from each other.

[dnj, 2017/03/08 04:13:43]

Done.

+        this.lastErrorValue = (err || null);
+
+        if (this.statusChangedCallback) {
+          this.statusChangedCallback();
+        };
+      }
+    }
+
+    /**
+     * Sets the status changed callback, which will be invoked whenever the
+     * Fetcher's status has changed.
+     */
+    setStatusChangedCallback(fn: () => void) {
+      this.statusChangedCallback = fn;
+    }
+
+    /**
+     * Returns a Promise that resolves to the next block of logs in the stream.

[hinoka, 2017/02/27 21:50:25]

"that will resolve to"

[dnj, 2017/03/08 04:13:43]

Done.

+     *
+     * @return {Promise[LogDog.LogEntry[]]} A Promise that will resolve to the
+     *     next block of logs in the stream, or null if there are no logs to

[hinoka, 2017/02/27 21:50:25]

"no logs to return currently" or "no more logs to return"?

[dnj, 2017/03/08 04:13:43]

Pretty sure this can't be null. Updated comment.

+     *     return.
+     */
+    get(index: number, opts: FetcherOptions): Promise<LogDog.LogEntry[]> {
+      return this.getIndex(index, opts);
+    }
+
+    /**
+     * Returns a Promise that resolves to "count" log entries starting at
+     * "startIndex".
+     *
+     * If multiple RPC calls are required to retrieve "count" entries, these
+     * will be scheduled, and the Promise will block until the full set of
+     * requested stream entries is retrieved.
+     */
+    getAll(startIndex: number, count: number): Promise<LogDog.LogEntry[]> {
+      // Request the tail walkback logs. Since our request for N logs may return
+      // <N logs, we will repeat the request until all requested logs have been
+      // obtained.
+      let allLogs: LogDog.LogEntry[] = [];
+
+      let getIter = (): Promise<LogDog.LogEntry[]> => {
+        if (count <= 0) {
+          return Promise.resolve(allLogs);
+        }
+
+        // Perform Gets until we have the requested number of logs. We don't
+        // have to constrain the "logCount" parameter b/c we automatically do
+        // that in getIndex.
+        let opts: FetcherOptions = {
+          logCount: count,
+          sparse: true,
+        };
+        return this.getIndex(startIndex, opts).then((logs) => {
+          if (logs) {
+            allLogs.push.apply(allLogs, logs);
+            startIndex += logs.length;
+            count -= logs.length;
+          }
+          return getIter();
+        });
+      };
+      return getIter();
+    }
+
+    /**
+     * Fetches the tail log entry.
+     */
+    tail(): Promise<LogDog.LogEntry[]> {

[hinoka, 2017/02/27 21:50:25]

getLast()?

[dnj, 2017/03/08 04:13:43]

getLatest

+      let streamingRetry = new luci.RetryIterator(Fetcher.streamingRetry);
+      let tryTail = (): Promise<LogDog.LogEntry[]> => {
+        return this.doTail().then((logs): Promise<LogDog.LogEntry[]> => {
+          if (logs && logs.length) {
+            return Promise.resolve(logs);
+          }
+
+          // No logs were returned, and we expect logs, so we're streaming. Try
+          // again after a delay.
+          this.setCurrentStatus(FetcherStatus.STREAMING);
+          let delay = streamingRetry.next();
+          console.warn(
+              this.stream, `: No logs returned; retrying after ${delay}ms...`);
+          return luci.sleepPromise(delay).then(() => {
+            return tryTail();
+          });
+        });
+      };
+      return tryTail();
+    }
+
+
+    private getIndex(index: number, opts: FetcherOptions):
+        Promise<LogDog.LogEntry[]> {
+      // (Testing) Constrain our max logs, if set.

[hinoka, 2017/02/27 21:50:25]

Is this actually testing?

[dnj, 2017/03/08 04:13:43]

Used by another module, but yes.

+      if (Fetcher.maxLogsPerGet > 0) {
+        if (!opts) {
+          opts = {};
+        }
+        if ((!opts.logCount) || opts.logCount > Fetcher.maxLogsPerGet) {
+          opts.logCount = Fetcher.maxLogsPerGet;
+        }
+      }
+
+      // We will retry continuously until we get a log (streaming).
+      let streamingRetry = new luci.RetryIterator(Fetcher.streamingRetry);

[hinoka, 2017/02/27 21:50:25]

Why are there 3 layers of retries (I counted 3.  the retryIterator(), the
recursive call below, and doRetryIfMissing)?  Can we just have one?

[dnj, 2017/03/08 04:13:43]

There should be two: "retryIterator":

streaming: the log exists, and is not terminated, but I haven't gotten all of
the things that I want, so try again since there could be more.

missing: the user said get this log, but LogDog says 404. I trust the user, this
is probably ingest lag, so try again.

Streaming backs off slowly to an upper bound until it gets more logs, then
resets to try again, backing off slowly.

Missing retries until a non-404 is returned.

The recursive call isn't a retry.

I've simplified this code by:
1) Adding a "do" method to a RetryIterator.
2) Using this everywhere here.

+      let tryGet = (): Promise<LogDog.LogEntry[]> => {
+        // If we're asking for a log beyond our stream, don't bother.
+        if (this.terminalIndex >= 0 && index > this.terminalIndex) {
+          return Promise.resolve([]);
+        }
+
+        return this.doGet(index, opts).then((logs) => {
+          if (logs && logs.length) {
+            // Since we allow non-contiguous Get, we may get back more logs than
+            // we actually expected. Prune any such additional.
+            if (opts.logCount > 0) {
+              let maxStreamIndex = index + opts.logCount - 1;
+              logs = logs.filter((le) => {
+                return le.streamIndex <= maxStreamIndex;
+              });
+            }
+
+            return Promise.resolve(logs);
+          }
+
+          // No logs were returned, and we expect logs, so we're streaming. Try
+          // again after a delay.
+          this.setCurrentStatus(FetcherStatus.STREAMING);
+          let delay = streamingRetry.next();
+          console.warn(
+              this.stream, `: No logs returned; retrying after ${delay}ms...`);
+          return luci.sleepPromise(delay).then(() => {
+            return tryGet();

[hinoka, 2017/02/27 21:50:25]

Wouldn't this increase the calling stack size on every retry?

[dnj, 2017/03/08 04:13:43]

I think so. Simplified via "do", which should not have this problem.

+          });
+        });
+      };
+      return tryGet();
+    }
+
+    private doGet(index: number, opts: FetcherOptions):
+        Promise<LogDog.LogEntry[]> {
+      let request: {
+        project: string; path: string; state: boolean; index: number;
+
+        nonContiguous?: boolean;
+        byteCount?: number;
+        logCount?: number;
+      } = {
+        project: this.stream.project,
+        path: this.stream.path,
+        state: (this.terminalIndex < 0),
+        index: index,
+      };
+      if (opts.sparse || this.archived) {
+        // This log stream is archived. We will relax the contiguous requirement
+        // so we can render sparse log streams.
+        request.nonContiguous = true;
+      }
+      if (opts) {
+        if (opts.byteCount > 0) {
+          request.byteCount = opts.byteCount;
+        }
+        if (opts.logCount > 0) {
+          request.logCount = opts.logCount;
+        }
+      }
+
+      if (this.debug) {
+        console.log('logdog.Logs.Get:', request);
+      }
+
+      // Perform our Get, waiting until the stream actually exists.
+      return this
+          .doRetryIfMissing(
+              ():
+                  Promise<FetchResult> => {
+                    return this.client.call('logdog.Logs', 'Get', request)
+                        .then((resp: GetResponse): FetchResult => {
+                          return FetchResult.make(resp, this.lastDesc);
+                        });
+                  })
+          .then((fr) => {
+            return this.afterProcessResult(fr);
+          });
+    }
+
+    private doTail(): Promise<LogDog.LogEntry[]> {

[hinoka, 2017/02/27 21:50:25]

doGetLast()?

[dnj, 2017/03/08 04:13:43]

Since Tail is actually the name of the RPC call, I'm going to leave this as-is.

+      let request: {project: string; path: string; state: boolean;} = {
+        project: this.stream.project,
+        path: this.stream.path,
+        state: (this.terminalIndex < 0),
+      };
+
+      if (this.debug) {
+        console.log('logdog.Logs.Tail:', request);
+      }
+
+      return this
+          .doRetryIfMissing(
+              ():
+                  Promise<FetchResult> => {
+                    return this.client.call('logdog.Logs', 'Tail', request)
+                        .then((resp: GetResponse): FetchResult => {
+                          return FetchResult.make(resp, this.lastDesc);
+                        });
+                  })
+          .then((fr) => {
+            return this.afterProcessResult(fr);
+          });
+    }
+
+    private afterProcessResult(fr: FetchResult): LogDog.LogEntry[] {
+      if (this.debug) {
+        if (fr.logs.length) {
+          console.log(
+              'Request returned:', fr.logs[0].streamIndex, '..',
+              fr.logs[fr.logs.length - 1].streamIndex, fr.desc, fr.state);
+        } else {
+          console.log('Request returned no logs:', fr.desc, fr.state);
+        }
+      }
+
+      this.setCurrentStatus(FetcherStatus.IDLE);
+      if (fr.desc) {
+        this.lastDesc = fr.desc;
+      }
+      if (fr.state) {
+        this.lastState = fr.state;
+      }
+      return fr.logs;
+    }
+
+    private doRetryIfMissing(fn: () => Promise<FetchResult>):
+        Promise<FetchResult> {
+      let missingRetry = new luci.RetryIterator(Fetcher.missingRetry);
+
+      let doIt = (): Promise<FetchResult> => {
+        this.setCurrentStatus(FetcherStatus.LOADING);
+
+        return fn().catch((err: Error) => {
+          // Is this a gRPC Error?
+          let grpc = luci.GrpcError.convert(err);
+          if (grpc && grpc.code === luci.Code.NOT_FOUND) {
+            this.setCurrentStatus(FetcherStatus.MISSING);
+
+            let delay = missingRetry.next();
+            console.warn(
+                this.stream, ': Is not found:', err,
+                `; retrying after ${delay}ms...`);
+            return luci.sleepPromise(delay).then(() => {
+              return doIt();
+            });
+          }
+
+          this.setCurrentStatus(FetcherStatus.ERROR, err);
+          throw err;
+        });
+      };
+      return doIt();
+    }
+  }
+
+  /**
+   * The result of a log stream fetch, for internal usage.
+   *
+   * It will include zero or more log entries, and optionally (if requested)
+   * the log stream's descriptor and state.
+   */
+  class FetchResult {
+    constructor(
+        readonly logs: LogDog.LogEntry[],
+        readonly desc?: LogDog.LogStreamDescriptor,
+        readonly state?: LogDog.LogStreamState) {}
+
+    static make(resp: GetResponse, desc: LogDog.LogStreamDescriptor):
+        FetchResult {
+      let loadDesc: LogDog.LogStreamDescriptor|undefined;
+      if (resp.desc) {
+        desc = loadDesc = LogDog.LogStreamDescriptor.make(resp.desc);
+      }
+
+      let loadState: LogDog.LogStreamState|undefined;
+      if (resp.state) {
+        loadState = LogDog.LogStreamState.make(resp.state);
+      }
+
+      let logs = (resp.logs || []).map((le) => {
+        return LogDog.LogEntry.make(le, desc);
+      });
+      return new FetchResult(logs, loadDesc, loadState);
+    }
+  }
+}