Bug 1677440 - Implement a ping uploader (#24)

Author: Beatriz Rizental

src/config.ts

@@ -0,0 +1,17 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+/**
+ * The Configuration class describes how to configure Glean.
+ */
+interface Configuration {
+  // The build identifier generated by the CI system (e.g. "1234/A").
+  readonly appBuild?: string,
+  // The user visible version string fro the application running Glean.js.
+  readonly appDisplayVersion?: string,
+  // The server pings are sent to.
+  readonly serverEndpoint?: URL
+}
+
+export default Configuration;

src/constants.ts

@@ -32,3 +32,6 @@ export const CLIENT_INFO_STORAGE = "glean_client_info";
 
 // We will set the client id to this client id in case upload is disabled.
 export const KNOWN_CLIENT_ID = "c0ffeec0-ffee-c0ff-eec0-ffeec0ffeec0";
+
+// The default server pings are sent to.
+export const DEFAULT_TELEMETRY_ENDPOINT = "https://incoming.telemetry.mozilla.org";

src/glean.ts

@@ -2,8 +2,11 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
+import { DEFAULT_TELEMETRY_ENDPOINT } from "./constants";
+import Configuration from "config";
 import MetricsDatabase from "metrics/database";
 import PingsDatabase from "pings/database";
+import PingUploader from "upload";
 import { isUndefined, sanitizeApplicationId } from "utils";
 import { CoreMetrics } from "internal_metrics";
 import { Lifetime } from "metrics";
@@ -23,9 +26,12 @@ class Glean {
   private _initialized: boolean;
   // Instances of Glean's core metrics.
   private _coreMetrics: CoreMetrics;
+  // The ping uploader.
+  private _pingUploader: PingUploader
 
   // Properties that will only be set on `initialize`.
   private _applicationId?: string;
+  private _serverEndpoint?: URL;
 
   private constructor() {
     if (!isUndefined(Glean._instance)) {
@@ -34,16 +40,21 @@ class Glean {
         Use Glean.instance instead to access the Glean singleton.`);
     }
 
+    this._pingUploader = new PingUploader();
     this._coreMetrics = new CoreMetrics();
     this._initialized = false;
     this._db = {
       metrics: new MetricsDatabase(),
-      pings: new PingsDatabase()
+      pings: new PingsDatabase(this._pingUploader)
     };
     // Temporarily setting this to true always, until Bug 1677444 is resolved.
     this._uploadEnabled = true;
   }
 
+  private static get pingUploader(): PingUploader {
+    return Glean.instance._pingUploader;
+  }
+
   private static get coreMetrics(): CoreMetrics {
     return Glean.instance._coreMetrics;
   }
@@ -60,19 +71,24 @@ class Glean {
    * Initialize Glean. This method should only be called once, subsequent calls will be no-op.
    *
    * @param applicationId The application ID (will be sanitized during initialization).
-   * @param appBuild The build identifier generated by the CI system (e.g. "1234/A").
-   * @param appDisplayVersion The user visible version string fro the application running Glean.js.
+   * @param config Glean configuration options.
    */
-  static async initialize(applicationId: string, appBuild?: string, appDisplayVersion?: string): Promise<void> {
+  static async initialize(applicationId: string, config?: Configuration): Promise<void> {
     if (Glean.instance._initialized) {
       console.warn("Attempted to initialize Glean, but it has already been initialized. Ignoring.");
       return;
     }
 
     Glean.instance._applicationId = sanitizeApplicationId(applicationId);
-    await Glean.coreMetrics.initialize(appBuild, appDisplayVersion);
+    Glean.instance._serverEndpoint = config
+      ? config.serverEndpoint : new URL(DEFAULT_TELEMETRY_ENDPOINT);
+    await Glean.coreMetrics.initialize(config?.appBuild, config?.appDisplayVersion);
 
     Glean.instance._initialized = true;
+
+    await Glean.pingUploader.scanPendingPings();
+    // Even though this returns a promise, there is no need to block on it returning.
+    Glean.pingUploader.triggerUpload();
   }
 
   /**
@@ -119,6 +135,14 @@ class Glean {
     return Glean.instance._applicationId;
   }
 
+  static get serverEndpoint(): URL | undefined {
+    if (!Glean.instance._initialized) {
+      console.error("Attempted to access the Glean.serverEndpoint before Glean was initialized.");
+    }
+
+    return Glean.instance._serverEndpoint;
+  }
+
   /**
    * **Test-only API**
    *
@@ -139,19 +163,23 @@ class Glean {
    * TODO: Only allow this function to be called on test mode (depends on Bug 1682771).
    *
    * @param applicationId The application ID (will be sanitized during initialization).
-   * @param optionalInitializeArgs Optional arguments to be passed to `initialize`.
+   * @param config Glean configuration options.
    */
-  static async testRestGlean(applicationId: string, ...optionalInitializeArgs: never[]): Promise<void> {
+  static async testRestGlean(applicationId: string, config?: Configuration): Promise<void> {
     // Get back to an uninitialized state.
     Glean.instance._initialized = false;
     // Reset upload enabled state, not to inerfere with other tests.
     Glean.uploadEnabled = true;
+
+    // Stop ongoing jobs and clear pending pings queue.
+    await Glean.pingUploader.clearPendingPingsQueue();
+
     // Clear the databases.
     await Glean.metricsDatabase.clearAll();
     await Glean.pingsDatabase.clearAll();
 
-    // Initialize Glean.
-    await Glean.initialize(applicationId, ...optionalInitializeArgs);
+    // Re-Initialize Glean.
+    await Glean.initialize(applicationId, config);
   }
 }
 

src/pings/database.ts

@@ -5,7 +5,7 @@
 import { Metrics as MetricsPayload } from "metrics/database";
 import { Store } from "storage";
 import PersistentStore from "storage/persistent";
-import { JSONObject } from "utils";
+import { isObject, isJSONValue, JSONObject, isString } from "utils";
 
 export interface PingInfo extends JSONObject {
   seq: number,
@@ -43,14 +43,46 @@ export interface PingPayload extends JSONObject {
   events?: JSONObject,
 }
 
+export interface PingInternalRepresentation extends JSONObject {
+  path: string,
+  payload: PingPayload,
+  headers?: Record<string, string>
+}
+
 /**
- * Debug headers to be added to a ping.
+ * Checks whether or not `v` is in the correct ping internal representation
+ *
+ * @param v The value to verify.
+ *
+ * @returns A special Typescript value (which compiles down to a boolean)
+ *          stating whether `v` is in the correct ping internal representation.
  */
-export interface PingHeaders extends JSONObject {
-  "X-Debug-Id"?: string,
-  "X-Source-Tags"?: string,
+export function isValidPingInternalRepresentation(v: unknown): v is PingInternalRepresentation {
+  if (isObject(v) && (Object.keys(v).length === 2 || Object.keys(v).length === 3)) {
+    const hasValidPath = "path" in v && isString(v.path);
+    const hasValidPayload = "payload" in v && isJSONValue(v.payload) && isObject(v.payload);
+    const hasValidHeaders = (!("headers" in v)) || (isJSONValue(v.headers) && isObject(v.headers));
+    if (!hasValidPath || !hasValidPayload || !hasValidHeaders) {
+      return false;
+    }
+    return true;
+  }
+  return false;
 }
 
+/**
+ * An interface to be implemented by classes that wish to observe the pings database.
+ */
+export interface Observer {
+  /**
+   * Updates an observer about a new ping of a given id
+   * that has just been recorded to the pings database.
+   *
+   * @param identifier The id of the ping that was just recorded.
+   * @param ping An object containing the newly recorded ping path, payload and optionally headers.
+   */
+  update(identifier: string, ping: PingInternalRepresentation): void;
+}
 
 /**
  * The pings database is an abstraction layer on top of the underlying storage.
@@ -67,9 +99,11 @@ export interface PingHeaders extends JSONObject {
  */
 class PingsDatabase {
   private store: Store;
+  private observer?: Observer;
 
-  constructor() {
+  constructor(observer?: Observer) {
     this.store = new PersistentStore("pings");
+    this.observer = observer;
   }
 
   /**
@@ -84,15 +118,51 @@ class PingsDatabase {
     path: string,
     identifier: string,
     payload: PingPayload,
-    headers?: PingHeaders
+    headers?: Record<string, string>
   ): Promise<void> {
-    await this.store.update([identifier], () => {
-      const base = {
-        path,
-        payload
-      };
-      return headers ? { ...base, headers } : base;
-    });
+    const ping: PingInternalRepresentation = {
+      path,
+      payload
+    };
+
+    if (headers) {
+      ping.headers = headers;
+    }
+
+    await this.store.update([identifier], () => ping);
+
+    // Notify the observer that a new ping has been added to the pings database.
+    this.observer && this.observer.update(identifier, ping);
+  }
+
+  /**
+   * Deletes a specific ping from the database.
+   *
+   * @param identifier The identififer of the ping to delete.
+   */
+  async deletePing(identifier: string): Promise<void> {
+    await this.store.delete([identifier]);
+  }
+
+  /**
+   * Gets all pings from the pings database. Deletes any data in unexpected format that is found.
+   *
+   * @returns List of all currently stored pings.
+   */
+  async getAllPings(): Promise<{ [id: string]: PingInternalRepresentation }> {
+    const allStoredPings = await this.store._getWholeStore();
+    const finalPings: { [ident: string]: PingInternalRepresentation } = {};
+    for (const identifier in allStoredPings) {
+      const ping = allStoredPings[identifier];
+      if (isValidPingInternalRepresentation(ping)) {
+        finalPings[identifier] = ping;
+      } else {
+        console.warn("Unexpected data found in pings database. Deleting.");
+        await this.store.delete([identifier]);
+      }
+    }
+
+    return finalPings;
   }
 
   /**

src/upload/adapter/index.ts

@@ -0,0 +1,28 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+/**
+ * Uploader interface, actualy uploading logic varies per platform.
+ */
+export interface UploadAdapter {
+  /**
+   * Makes a POST request to a given url, with the given headers and body.
+   *
+   * @param url The URL to make the POST request
+   * @param body The stringified body of this post request
+   * @param headers Optional header to include in the request
+   *
+   * @returns The status code of the response.
+   */
+  post(url: URL, body: string, headers?: Record<string, string>): Promise<number>;
+}
+
+// Default export for tests sake.
+const MockUploadAdapter: UploadAdapter = {
+  post(): Promise<number> {
+    return Promise.resolve(200);
+  }
+};
+
+export default MockUploadAdapter;