| OLD | NEW |
|---|
| (Empty) |
| 1 <!-- | |
| 2 Copyright 2016 The LUCI Authors. All rights reserved. | |
| 3 Use of this source code is governed under the Apache License, Version 2.0 | |
| 4 that can be found in the LICENSE file. | |
| 5 --> | |
| 6 | |
| 7 <link rel="import" href="../bower_components/polymer/polymer.html"> | |
| 8 <link rel="import" href="../bower_components/promise-polyfill/promise-polyfill-l
ite.html"> | |
| 9 | |
| 10 <script> | |
| 11 "use strict"; | |
| 12 | |
| 13 function LogDogFetcher(client, stream) { | |
| 14 this.client = client; | |
| 15 this.stream = stream; | |
| 16 | |
| 17 // Fetching parameters, will be updated as logs are fetched. | |
| 18 this.sleepTimeSecs = 5; | |
| 19 this.byteCount = null; | |
| 20 this.logCount = null; | |
| 21 this.reset(); | |
| 22 } | |
| 23 | |
| 24 LogDogFetcher.prototype.reset = function() { | |
| 25 this.nextIndex = 0; | |
| 26 this.finished = false; | |
| 27 this.desc = null; | |
| 28 this.state = null; | |
| 29 | |
| 30 this._current = null; | |
| 31 this._nextLogsPromise = null; | |
| 32 }; | |
| 33 | |
| 34 /** | |
| 35 * Returns the log stream's terminal index. | |
| 36 * | |
| 37 * If no terminal index is known, or if the log stream is still streaming, | |
| 38 * this will return -1. | |
| 39 */ | |
| 40 LogDogFetcher.prototype.terminalIndex = function() { | |
| 41 return (this.state) ? (this.state.terminalIndex) : (-1); | |
| 42 }; | |
| 43 | |
| 44 /** | |
| 45 * Returns a Promise that resolves to the next block of logs in the stream. | |
| 46 * | |
| 47 * If there are no more logs in the stream (finished), the returned Promise | |
| 48 * will already be resolved and will contain a null log. | |
| 49 * | |
| 50 * @return {Promise[Object]} A Promise that will resolve to the next block | |
| 51 * of logs in the stream. | |
| 52 */ | |
| 53 LogDogFetcher.prototype.next = function() { | |
| 54 // If we don't have an in-progress fetch, start a new one. | |
| 55 if (this._nextLogsPromise === null) { | |
| 56 this._nextLogsPromise = this._fetchNextBatch(). | |
| 57 then(function(result) { | |
| 58 var entries = result.entries; | |
| 59 if (entries && entries.length) { | |
| 60 var lastIndex = entries[entries.length-1].streamIndex; | |
| 61 this.nextIndex = (lastIndex + 1); | |
| 62 | |
| 63 var tidx = this.terminalIndex(); | |
| 64 if (tidx >= 0 && tidx < this.nextIndex) { | |
| 65 // We have punted the full log stream. Mark finished. | |
| 66 this.finished = true; | |
| 67 } | |
| 68 } | |
| 69 | |
| 70 this._nextLogsPromise = null; | |
| 71 return result; | |
| 72 }.bind(this)).catch(function(error) { | |
| 73 this._nextLogsPromise = null; | |
| 74 throw error; | |
| 75 }.bind(this)); | |
| 76 } | |
| 77 return this._nextLogsPromise; | |
| 78 }, | |
| 79 | |
| 80 /** Creates and returns a Promise for the next batch of logs. */ | |
| 81 LogDogFetcher.prototype._fetchNextBatch = function() { | |
| 82 // If we're already finished, return the terminal result. | |
| 83 if (this.finished) { | |
| 84 return this._resolvedLogs(null); | |
| 85 } | |
| 86 | |
| 87 // Fetch and return the next batch of logs. | |
| 88 return this._scheduleAsyncGet().then(function(resp) { | |
| 89 // Update our state/desc. | |
| 90 if (resp.state) { | |
| 91 this.state = resp.state; | |
| 92 } | |
| 93 if (resp.desc) { | |
| 94 this.desc = resp.desc; | |
| 95 } | |
| 96 | |
| 97 var logs = resp.logs; | |
| 98 if (! (logs && logs.length)) { | |
| 99 // No logs were loaded this round. Sleep for a bit then try again. | |
| 100 // (Streaming case). | |
| 101 console.log("No logs for", this.stream, "; sleeping..."); | |
| 102 return new LuciSleepPromise(this.sleepTimeSecs * 1000). | |
| 103 then(function() { | |
| 104 return this._fetchNextBatch(); | |
| 105 }.bind(this)); | |
| 106 } | |
| 107 | |
| 108 return this._resolvedLogs(logs); | |
| 109 }.bind(this)); | |
| 110 }; | |
| 111 | |
| 112 /** Generates a structured Promise for a given block of log entries. */ | |
| 113 LogDogFetcher.prototype._resolvedLogs = function(punt) { | |
| 114 return Promise.resolve({ | |
| 115 desc: this.desc, | |
| 116 state: this.state, | |
| 117 entries: punt, | |
| 118 }); | |
| 119 }; | |
| 120 | |
| 121 /** Schedules the next asynchronous fetch. */ | |
| 122 LogDogFetcher.prototype._scheduleAsyncGet = function() { | |
| 123 this.client.service = "logdog.Logs"; | |
| 124 this.client.method = "Get"; | |
| 125 this.client.request = { | |
| 126 project: this.stream.project, | |
| 127 path: this.stream.path, | |
| 128 state: (!this.state || this.terminalIndex() < 0), | |
| 129 index: this.nextIndex, | |
| 130 }; | |
| 131 | |
| 132 if (this.byteCount !== null) { | |
| 133 this.client.request.byteCount = this.byteCount; | |
| 134 } | |
| 135 if (this.logCount !== null) { | |
| 136 this.client.request.logCount = this.logCount; | |
| 137 } | |
| 138 | |
| 139 return this.client.call().completes.then(function(resp) { | |
| 140 resp = resp.response; | |
| 141 | |
| 142 // Normalize the resulting logs. | |
| 143 // | |
| 144 // JSONPB timestamps are in the form of RFC3339 strings. | |
| 145 if (resp.desc) { | |
| 146 patchDescriptor(resp.desc); | |
| 147 } | |
| 148 if (resp.state) { | |
| 149 patchState(resp.state); | |
| 150 } | |
| 151 if (resp.logs) { | |
| 152 resp.logs.forEach(function(le) { | |
| 153 patchLogEntry(le, resp.desc); | |
| 154 }); | |
| 155 } | |
| 156 | |
| 157 return resp; | |
| 158 }, function(error) { | |
| 159 throw LogDogError.wrapGrpc(error); | |
| 160 }); | |
| 161 }; | |
| 162 </script> | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 2717043002:
Add LogDog log stream fetcher code. (Closed)
