summaryrefslogtreecommitdiffstats
path: root/services/sync/modules/telemetry.js
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--services/sync/modules/telemetry.js1111
1 files changed, 1111 insertions, 0 deletions
diff --git a/services/sync/modules/telemetry.js b/services/sync/modules/telemetry.js
new file mode 100644
index 0000000000..12150ade91
--- /dev/null
+++ b/services/sync/modules/telemetry.js
@@ -0,0 +1,1111 @@
+/* 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/. */
+
+"use strict";
+
+var EXPORTED_SYMBOLS = ["SyncTelemetry"];
+
+// Support for Sync-and-FxA-related telemetry, which is submitted in a special-purpose
+// telemetry ping called the "sync ping", documented here:
+//
+// ../../../toolkit/components/telemetry/docs/data/sync-ping.rst
+//
+// The sync ping contains identifiers that are linked to the user's Firefox Account
+// and are separate from the main telemetry client_id, so this file is also responsible
+// for ensuring that we can delete those pings upon user request, by plumbing its
+// identifiers into the "deletion-request" ping.
+
+const { XPCOMUtils } = ChromeUtils.import(
+ "resource://gre/modules/XPCOMUtils.jsm"
+);
+
+XPCOMUtils.defineLazyModuleGetters(this, {
+ Async: "resource://services-common/async.js",
+ AuthenticationError: "resource://services-sync/browserid_identity.js",
+ fxAccounts: "resource://gre/modules/FxAccounts.jsm",
+ FxAccounts: "resource://gre/modules/FxAccounts.jsm",
+ Log: "resource://gre/modules/Log.jsm",
+ ObjectUtils: "resource://gre/modules/ObjectUtils.jsm",
+ Observers: "resource://services-common/observers.js",
+ OS: "resource://gre/modules/osfile.jsm",
+ Resource: "resource://services-sync/resource.js",
+ Services: "resource://gre/modules/Services.jsm",
+ Status: "resource://services-sync/status.js",
+ Svc: "resource://services-sync/util.js",
+ TelemetryController: "resource://gre/modules/TelemetryController.jsm",
+ TelemetryEnvironment: "resource://gre/modules/TelemetryEnvironment.jsm",
+ TelemetryUtils: "resource://gre/modules/TelemetryUtils.jsm",
+ Weave: "resource://services-sync/main.js",
+});
+
+let constants = {};
+ChromeUtils.import("resource://services-sync/constants.js", constants);
+
+XPCOMUtils.defineLazyServiceGetter(
+ this,
+ "Telemetry",
+ "@mozilla.org/base/telemetry;1",
+ "nsITelemetry"
+);
+
+XPCOMUtils.defineLazyGetter(
+ this,
+ "WeaveService",
+ () => Cc["@mozilla.org/weave/service;1"].getService().wrappedJSObject
+);
+const log = Log.repository.getLogger("Sync.Telemetry");
+
+const TOPICS = [
+ // For tracking change to account/device identifiers.
+ "fxaccounts:new_device_id",
+ "fxaccounts:onlogout",
+ "weave:service:ready",
+ "weave:service:login:change",
+
+ // For whole-of-sync metrics.
+ "weave:service:sync:start",
+ "weave:service:sync:finish",
+ "weave:service:sync:error",
+
+ // For individual engine metrics.
+ "weave:engine:sync:start",
+ "weave:engine:sync:finish",
+ "weave:engine:sync:error",
+ "weave:engine:sync:applied",
+ "weave:engine:sync:step",
+ "weave:engine:sync:uploaded",
+ "weave:engine:validate:finish",
+ "weave:engine:validate:error",
+
+ // For ad-hoc telemetry events.
+ "weave:telemetry:event",
+ "weave:telemetry:histogram",
+ "fxa:telemetry:event",
+
+ "weave:telemetry:migration",
+];
+
+const PING_FORMAT_VERSION = 1;
+
+const EMPTY_UID = "0".repeat(32);
+
+// The set of engines we record telemetry for - any other engines are ignored.
+const ENGINES = new Set([
+ "addons",
+ "bookmarks",
+ "clients",
+ "forms",
+ "history",
+ "passwords",
+ "prefs",
+ "tabs",
+ "extension-storage",
+ "addresses",
+ "creditcards",
+]);
+
+// A regex we can use to replace the profile dir in error messages. We use a
+// regexp so we can simply replace all case-insensitive occurences.
+// This escaping function is from:
+// https://developer.mozilla.org/en/docs/Web/JavaScript/Guide/Regular_Expressions
+const reProfileDir = new RegExp(
+ OS.Constants.Path.profileDir.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"),
+ "gi"
+);
+
+function tryGetMonotonicTimestamp() {
+ try {
+ return Telemetry.msSinceProcessStart();
+ } catch (e) {
+ log.warn("Unable to get a monotonic timestamp!");
+ return -1;
+ }
+}
+
+function timeDeltaFrom(monotonicStartTime) {
+ let now = tryGetMonotonicTimestamp();
+ if (monotonicStartTime !== -1 && now !== -1) {
+ return Math.round(now - monotonicStartTime);
+ }
+ return -1;
+}
+
+// Converts extra integer fields to strings, rounds floats to three
+// decimal places (nanosecond precision for timings), and removes profile
+// directory paths and URLs from potential error messages.
+function normalizeExtraTelemetryFields(extra) {
+ let result = {};
+ for (let key in extra) {
+ let value = extra[key];
+ let type = typeof value;
+ if (type == "string") {
+ result[key] = cleanErrorMessage(value);
+ } else if (type == "number") {
+ result[key] = Number.isInteger(value)
+ ? value.toString(10)
+ : value.toFixed(3);
+ } else if (type != "undefined") {
+ throw new TypeError(
+ `Invalid type ${type} for extra telemetry field ${key}`
+ );
+ }
+ }
+ return ObjectUtils.isEmpty(result) ? undefined : result;
+}
+
+// This function validates the payload of a telemetry "event" - this can be
+// removed once there are APIs available for the telemetry modules to collect
+// these events (bug 1329530) - but for now we simulate that planned API as
+// best we can.
+function validateTelemetryEvent(eventDetails) {
+ let { object, method, value, extra } = eventDetails;
+ // Do do basic validation of the params - everything except "extra" must
+ // be a string. method and object are required.
+ if (
+ typeof method != "string" ||
+ typeof object != "string" ||
+ (value && typeof value != "string") ||
+ (extra && typeof extra != "object")
+ ) {
+ log.warn("Invalid event parameters - wrong types", eventDetails);
+ return false;
+ }
+ // length checks.
+ if (
+ method.length > 20 ||
+ object.length > 20 ||
+ (value && value.length > 80)
+ ) {
+ log.warn("Invalid event parameters - wrong lengths", eventDetails);
+ return false;
+ }
+
+ // extra can be falsey, or an object with string names and values.
+ if (extra) {
+ if (Object.keys(extra).length > 10) {
+ log.warn("Invalid event parameters - too many extra keys", eventDetails);
+ return false;
+ }
+ for (let [ename, evalue] of Object.entries(extra)) {
+ if (
+ typeof ename != "string" ||
+ ename.length > 15 ||
+ typeof evalue != "string" ||
+ evalue.length > 85
+ ) {
+ log.warn(
+ `Invalid event parameters: extra item "${ename} is invalid`,
+ eventDetails
+ );
+ return false;
+ }
+ }
+ }
+ return true;
+}
+
+class EngineRecord {
+ constructor(name) {
+ // startTime is in ms from process start, but is monotonic (unlike Date.now())
+ // so we need to keep both it and when.
+ this.startTime = tryGetMonotonicTimestamp();
+ this.name = name;
+
+ // This allows cases like bookmarks-buffered to have a separate name from
+ // the bookmarks engine.
+ let engineImpl = Weave.Service.engineManager.get(name);
+ if (engineImpl && engineImpl.overrideTelemetryName) {
+ this.overrideTelemetryName = engineImpl.overrideTelemetryName;
+ }
+ }
+
+ toJSON() {
+ let result = { name: this.overrideTelemetryName || this.name };
+ let properties = [
+ "took",
+ "status",
+ "failureReason",
+ "incoming",
+ "outgoing",
+ "validation",
+ "steps",
+ ];
+ for (let property of properties) {
+ result[property] = this[property];
+ }
+ return result;
+ }
+
+ finished(error) {
+ let took = timeDeltaFrom(this.startTime);
+ if (took > 0) {
+ this.took = took;
+ }
+ if (error) {
+ this.failureReason = SyncTelemetry.transformError(error);
+ }
+ }
+
+ recordApplied(counts) {
+ if (this.incoming) {
+ log.error(
+ `Incoming records applied multiple times for engine ${this.name}!`
+ );
+ return;
+ }
+ if (this.name === "clients" && !counts.failed) {
+ // ignore successful application of client records
+ // since otherwise they show up every time and are meaningless.
+ return;
+ }
+
+ let incomingData = {};
+ let properties = ["applied", "failed", "newFailed", "reconciled"];
+ // Only record non-zero properties and only record incoming at all if
+ // there's at least one property we care about.
+ for (let property of properties) {
+ if (counts[property]) {
+ incomingData[property] = counts[property];
+ this.incoming = incomingData;
+ }
+ }
+ }
+
+ recordStep(stepResult) {
+ let step = {
+ name: stepResult.name,
+ };
+ if (stepResult.took > 0) {
+ step.took = Math.round(stepResult.took);
+ }
+ if (stepResult.counts) {
+ let counts = stepResult.counts.filter(({ count }) => count > 0);
+ if (counts.length) {
+ step.counts = counts;
+ }
+ }
+ if (this.steps) {
+ this.steps.push(step);
+ } else {
+ this.steps = [step];
+ }
+ }
+
+ recordValidation(validationResult) {
+ if (this.validation) {
+ log.error(`Multiple validations occurred for engine ${this.name}!`);
+ return;
+ }
+ let { problems, version, took, checked } = validationResult;
+ let validation = {
+ version: version || 0,
+ checked: checked || 0,
+ };
+ if (took > 0) {
+ validation.took = Math.round(took);
+ }
+ let summarized = problems.filter(({ count }) => count > 0);
+ if (summarized.length) {
+ validation.problems = summarized;
+ }
+ this.validation = validation;
+ }
+
+ recordValidationError(e) {
+ if (this.validation) {
+ log.error(`Multiple validations occurred for engine ${this.name}!`);
+ return;
+ }
+
+ this.validation = {
+ failureReason: SyncTelemetry.transformError(e),
+ };
+ }
+
+ recordUploaded(counts) {
+ if (counts.sent || counts.failed) {
+ if (!this.outgoing) {
+ this.outgoing = [];
+ }
+ this.outgoing.push({
+ sent: counts.sent || undefined,
+ failed: counts.failed || undefined,
+ });
+ }
+ }
+}
+
+// The record of a single "sync" - typically many of these are submitted in
+// a single ping (ie, as a 'syncs' array)
+class SyncRecord {
+ constructor(allowedEngines, why) {
+ this.allowedEngines = allowedEngines;
+ // Our failure reason. This property only exists in the generated ping if an
+ // error actually occurred.
+ this.failureReason = undefined;
+ this.syncNodeType = null;
+ this.when = Date.now();
+ this.startTime = tryGetMonotonicTimestamp();
+ this.took = 0; // will be set later.
+ this.why = why;
+
+ // All engines that have finished (ie, does not include the "current" one)
+ // We omit this from the ping if it's empty.
+ this.engines = [];
+ // The engine that has started but not yet stopped.
+ this.currentEngine = null;
+ }
+
+ toJSON() {
+ let result = {
+ when: this.when,
+ took: this.took,
+ failureReason: this.failureReason,
+ status: this.status,
+ };
+ if (this.why) {
+ result.why = this.why;
+ }
+ let engines = [];
+ for (let engine of this.engines) {
+ engines.push(engine.toJSON());
+ }
+ if (engines.length > 0) {
+ result.engines = engines;
+ }
+ return result;
+ }
+
+ finished(error) {
+ this.took = timeDeltaFrom(this.startTime);
+ if (this.currentEngine != null) {
+ log.error(
+ "Finished called for the sync before the current engine finished"
+ );
+ this.currentEngine.finished(null);
+ this.onEngineStop(this.currentEngine.name);
+ }
+ if (error) {
+ this.failureReason = SyncTelemetry.transformError(error);
+ }
+
+ this.syncNodeType = Weave.Service.identity.telemetryNodeType;
+
+ // Check for engine statuses. -- We do this now, and not in engine.finished
+ // to make sure any statuses that get set "late" are recorded
+ for (let engine of this.engines) {
+ let status = Status.engines[engine.name];
+ if (status && status !== constants.ENGINE_SUCCEEDED) {
+ engine.status = status;
+ }
+ }
+
+ let statusObject = {};
+
+ let serviceStatus = Status.service;
+ if (serviceStatus && serviceStatus !== constants.STATUS_OK) {
+ statusObject.service = serviceStatus;
+ this.status = statusObject;
+ }
+ let syncStatus = Status.sync;
+ if (syncStatus && syncStatus !== constants.SYNC_SUCCEEDED) {
+ statusObject.sync = syncStatus;
+ this.status = statusObject;
+ }
+ }
+
+ onEngineStart(engineName) {
+ if (this._shouldIgnoreEngine(engineName, false)) {
+ return;
+ }
+
+ if (this.currentEngine) {
+ log.error(
+ `Being told that engine ${engineName} has started, but current engine ${this.currentEngine.name} hasn't stopped`
+ );
+ // Just discard the current engine rather than making up data for it.
+ }
+ this.currentEngine = new EngineRecord(engineName);
+ }
+
+ onEngineStop(engineName, error) {
+ // We only care if it's the current engine if we have a current engine.
+ if (this._shouldIgnoreEngine(engineName, !!this.currentEngine)) {
+ return;
+ }
+ if (!this.currentEngine) {
+ // It's possible for us to get an error before the start message of an engine
+ // (somehow), in which case we still want to record that error.
+ if (!error) {
+ return;
+ }
+ log.error(
+ `Error triggered on ${engineName} when no current engine exists: ${error}`
+ );
+ this.currentEngine = new EngineRecord(engineName);
+ }
+ this.currentEngine.finished(error);
+ this.engines.push(this.currentEngine);
+ this.currentEngine = null;
+ }
+
+ onEngineApplied(engineName, counts) {
+ if (this._shouldIgnoreEngine(engineName)) {
+ return;
+ }
+ this.currentEngine.recordApplied(counts);
+ }
+
+ onEngineStep(engineName, step) {
+ if (this._shouldIgnoreEngine(engineName)) {
+ return;
+ }
+ this.currentEngine.recordStep(step);
+ }
+
+ onEngineValidated(engineName, validationData) {
+ if (this._shouldIgnoreEngine(engineName, false)) {
+ return;
+ }
+ let engine = this.engines.find(e => e.name === engineName);
+ if (
+ !engine &&
+ this.currentEngine &&
+ engineName === this.currentEngine.name
+ ) {
+ engine = this.currentEngine;
+ }
+ if (engine) {
+ engine.recordValidation(validationData);
+ } else {
+ log.warn(
+ `Validation event triggered for engine ${engineName}, which hasn't been synced!`
+ );
+ }
+ }
+
+ onEngineValidateError(engineName, error) {
+ if (this._shouldIgnoreEngine(engineName, false)) {
+ return;
+ }
+ let engine = this.engines.find(e => e.name === engineName);
+ if (
+ !engine &&
+ this.currentEngine &&
+ engineName === this.currentEngine.name
+ ) {
+ engine = this.currentEngine;
+ }
+ if (engine) {
+ engine.recordValidationError(error);
+ } else {
+ log.warn(
+ `Validation failure event triggered for engine ${engineName}, which hasn't been synced!`
+ );
+ }
+ }
+
+ onEngineUploaded(engineName, counts) {
+ if (this._shouldIgnoreEngine(engineName)) {
+ return;
+ }
+ this.currentEngine.recordUploaded(counts);
+ }
+
+ _shouldIgnoreEngine(engineName, shouldBeCurrent = true) {
+ if (!this.allowedEngines.has(engineName)) {
+ log.info(
+ `Notification for engine ${engineName}, but we aren't recording telemetry for it`
+ );
+ return true;
+ }
+ if (shouldBeCurrent) {
+ if (!this.currentEngine || engineName != this.currentEngine.name) {
+ log.error(`Notification for engine ${engineName} but it isn't current`);
+ return true;
+ }
+ }
+ return false;
+ }
+}
+
+function cleanErrorMessage(error) {
+ // There's a chance the profiledir is in the error string which is PII we
+ // want to avoid including in the ping.
+ error = error.replace(reProfileDir, "[profileDir]");
+ // MSG_INVALID_URL from /dom/bindings/Errors.msg -- no way to access this
+ // directly from JS.
+ if (error.endsWith("is not a valid URL.")) {
+ error = "<URL> is not a valid URL.";
+ }
+ // Try to filter things that look somewhat like a URL (in that they contain a
+ // colon in the middle of non-whitespace), in case anything else is including
+ // these in error messages. Note that JSON.stringified stuff comes through
+ // here, so we explicitly ignore double-quotes as well.
+ error = error.replace(/[^\s"]+:[^\s"]+/g, "<URL>");
+ return error;
+}
+
+// The entire "sync ping" - it includes all the syncs, events etc recorded in
+// the ping.
+class SyncTelemetryImpl {
+ constructor(allowedEngines) {
+ log.manageLevelFromPref("services.sync.log.logger.telemetry");
+ // This is accessible so we can enable custom engines during tests.
+ this.allowedEngines = allowedEngines;
+ this.current = null;
+ this.setupObservers();
+
+ this.payloads = [];
+ this.discarded = 0;
+ this.events = [];
+ this.histograms = {};
+ this.migrations = [];
+ this.maxEventsCount = Svc.Prefs.get("telemetry.maxEventsCount", 1000);
+ this.maxPayloadCount = Svc.Prefs.get("telemetry.maxPayloadCount");
+ this.submissionInterval =
+ Svc.Prefs.get("telemetry.submissionInterval") * 1000;
+ this.lastSubmissionTime = Telemetry.msSinceProcessStart();
+ this.lastUID = EMPTY_UID;
+ this.lastSyncNodeType = null;
+ this.currentSyncNodeType = null;
+ // Note that the sessionStartDate is somewhat arbitrary - the telemetry
+ // modules themselves just use `new Date()`. This means that our startDate
+ // isn't going to be the same as the sessionStartDate in the main pings,
+ // but that's OK for now - if it's a problem we'd need to change the
+ // telemetry modules to expose what it thinks the sessionStartDate is.
+ let sessionStartDate = new Date();
+ this.sessionStartDate = TelemetryUtils.toLocalTimeISOString(
+ TelemetryUtils.truncateToHours(sessionStartDate)
+ );
+ TelemetryController.registerSyncPingShutdown(() => this.shutdown());
+ }
+
+ sanitizeFxaDeviceId(deviceId) {
+ return fxAccounts.telemetry.sanitizeDeviceId(deviceId);
+ }
+
+ prepareFxaDevices(devices) {
+ // For non-sync users, the data per device is limited -- just an id and a
+ // type (and not even the id yet). For sync users, if we can correctly map
+ // the fxaDevice to a sync device, then we can get os and version info,
+ // which would be quite unfortunate to lose.
+ let extraInfoMap = new Map();
+ if (this.syncIsEnabled()) {
+ for (let client of this.getClientsEngineRecords()) {
+ if (client.fxaDeviceId) {
+ extraInfoMap.set(client.fxaDeviceId, {
+ os: client.os,
+ version: client.version,
+ syncID: this.sanitizeFxaDeviceId(client.id),
+ });
+ }
+ }
+ }
+ // Finally, sanitize and convert to the proper format.
+ return devices.map(d => {
+ let { os, version, syncID } = extraInfoMap.get(d.id) || {
+ os: undefined,
+ version: undefined,
+ syncID: undefined,
+ };
+ return {
+ id: this.sanitizeFxaDeviceId(d.id) || EMPTY_UID,
+ type: d.type,
+ os,
+ version,
+ syncID,
+ };
+ });
+ }
+
+ syncIsEnabled() {
+ return WeaveService.enabled && WeaveService.ready;
+ }
+
+ // Separate for testing.
+ getClientsEngineRecords() {
+ if (!this.syncIsEnabled()) {
+ throw new Error("Bug: syncIsEnabled() must be true, check it first");
+ }
+ return Weave.Service.clientsEngine.remoteClients;
+ }
+
+ updateFxaDevices(devices) {
+ if (!devices) {
+ return {};
+ }
+ let me = devices.find(d => d.isCurrentDevice);
+ let id = me ? this.sanitizeFxaDeviceId(me.id) : undefined;
+ let cleanDevices = this.prepareFxaDevices(devices);
+ return { deviceID: id, devices: cleanDevices };
+ }
+
+ getFxaDevices() {
+ return fxAccounts.device.recentDeviceList;
+ }
+
+ getPingJSON(reason) {
+ let { devices, deviceID } = this.updateFxaDevices(this.getFxaDevices());
+ return {
+ os: TelemetryEnvironment.currentEnvironment.system.os,
+ why: reason,
+ devices,
+ discarded: this.discarded || undefined,
+ version: PING_FORMAT_VERSION,
+ syncs: this.payloads.slice(),
+ uid: this.lastUID,
+ syncNodeType: this.lastSyncNodeType || undefined,
+ deviceID,
+ sessionStartDate: this.sessionStartDate,
+ events: this.events.length == 0 ? undefined : this.events,
+ migrations: this.migrations.length == 0 ? undefined : this.migrations,
+ histograms:
+ Object.keys(this.histograms).length == 0 ? undefined : this.histograms,
+ };
+ }
+
+ _addMigrationRecord(type, info) {
+ log.debug("Saw telemetry migration info", type, info);
+ // Updates to this need to be documented in `sync-ping.rst`
+ switch (type) {
+ case "webext-storage":
+ this.migrations.push({
+ type: "webext-storage",
+ entries: +info.entries,
+ entriesSuccessful: +info.entries_successful,
+ extensions: +info.extensions,
+ extensionsSuccessful: +info.extensions_successful,
+ openFailure: !!info.open_failure,
+ });
+ break;
+ default:
+ throw new Error("Bug: Unknown migration record type " + type);
+ }
+ }
+
+ finish(reason) {
+ // Note that we might be in the middle of a sync right now, and so we don't
+ // want to touch this.current.
+ let result = this.getPingJSON(reason);
+ this.payloads = [];
+ this.discarded = 0;
+ this.events = [];
+ this.migrations = [];
+ this.histograms = {};
+ this.submit(result);
+ }
+
+ setupObservers() {
+ for (let topic of TOPICS) {
+ Observers.add(topic, this, this);
+ }
+ }
+
+ shutdown() {
+ this.finish("shutdown");
+ for (let topic of TOPICS) {
+ Observers.remove(topic, this, this);
+ }
+ }
+
+ submit(record) {
+ if (!this.isProductionSyncUser()) {
+ return false;
+ }
+ // We still call submit() with possibly illegal payloads so that tests can
+ // know that the ping was built. We don't end up submitting them, however.
+ let numEvents = record.events ? record.events.length : 0;
+ let numMigrations = record.migrations ? record.migrations.length : 0;
+ if (record.syncs.length || numEvents || numMigrations) {
+ log.trace(
+ `submitting ${record.syncs.length} sync record(s) and ` +
+ `${numEvents} event(s) to telemetry`
+ );
+ TelemetryController.submitExternalPing("sync", record, {
+ usePingSender: true,
+ }).catch(err => {
+ log.error("failed to submit ping", err);
+ });
+ return true;
+ }
+ return false;
+ }
+
+ isProductionSyncUser() {
+ // If FxA isn't production then we treat sync as not being production.
+ // Further, there's the deprecated "services.sync.tokenServerURI" pref we
+ // need to consider - fxa doesn't consider that as if that's the only
+ // pref set, they *are* running a production fxa, just not production sync.
+ if (
+ !FxAccounts.config.isProductionConfig() ||
+ Services.prefs.prefHasUserValue("services.sync.tokenServerURI")
+ ) {
+ log.trace(`Not sending telemetry ping for self-hosted Sync user`);
+ return false;
+ }
+ return true;
+ }
+
+ onSyncStarted(data) {
+ const why = data && JSON.parse(data).why;
+ if (this.current) {
+ log.warn(
+ "Observed weave:service:sync:start, but we're already recording a sync!"
+ );
+ // Just discard the old record, consistent with our handling of engines, above.
+ this.current = null;
+ }
+ this.current = new SyncRecord(this.allowedEngines, why);
+ }
+
+ // We need to ensure that the telemetry `deletion-request` ping always contains the user's
+ // current sync device ID, because if the user opts out of telemetry then the deletion ping
+ // will be immediately triggered for sending, and we won't have a chance to fill it in later.
+ // This keeps the `deletion-ping` up-to-date when the user's account state changes.
+ onAccountInitOrChange() {
+ // We don't submit sync pings for self-hosters, so don't need to collect their device ids either.
+ if (!this.isProductionSyncUser()) {
+ return;
+ }
+ // Awkwardly async, but no need to await. If the user's account state changes while
+ // this promise is in flight, it will reject and we won't record any data in the ping.
+ // (And a new notification will trigger us to try again with the new state).
+ fxAccounts.device
+ .getLocalId()
+ .then(deviceId => {
+ let sanitizedDeviceId = fxAccounts.telemetry.sanitizeDeviceId(deviceId);
+ // In the past we did not persist the FxA metrics identifiers to disk,
+ // so this might be missing until we can fetch it from the server for the
+ // first time. There will be a fresh notification tirggered when it's available.
+ if (sanitizedDeviceId) {
+ // Sanitized device ids are 64 characters long, but telemetry limits scalar strings to 50.
+ // The first 32 chars are sufficient to uniquely identify the device, so just send those.
+ // It's hard to change the sync ping itself to only send 32 chars, to b/w compat reasons.
+ sanitizedDeviceId = sanitizedDeviceId.substr(0, 32);
+ Services.telemetry.scalarSet(
+ "deletion.request.sync_device_id",
+ sanitizedDeviceId
+ );
+ }
+ })
+ .catch(err => {
+ log.warn(
+ `Failed to set sync identifiers in the deletion-request ping: ${err}`
+ );
+ });
+ }
+
+ // This keeps the `deletion-request` ping up-to-date when the user signs out,
+ // clearing the now-nonexistent sync device id.
+ onAccountLogout() {
+ Services.telemetry.scalarSet("deletion.request.sync_device_id", "");
+ }
+
+ _checkCurrent(topic) {
+ if (!this.current) {
+ log.warn(
+ `Observed notification ${topic} but no current sync is being recorded.`
+ );
+ return false;
+ }
+ return true;
+ }
+
+ _shouldSubmitForDataChange() {
+ let newID = fxAccounts.telemetry.getSanitizedUID() || EMPTY_UID;
+ let oldID = this.lastUID;
+ if (
+ newID != EMPTY_UID &&
+ oldID != EMPTY_UID &&
+ // Both are "real" uids, so we care if they've changed.
+ newID != oldID
+ ) {
+ log.trace(
+ `shouldSubmitForDataChange - uid from '${oldID}' -> '${newID}'`
+ );
+ return true;
+ }
+ // We've gone from knowing one of the ids to not knowing it (which we
+ // ignore) or we've gone from not knowing it to knowing it (which is fine),
+ // Now check the node type because a change there also means we should
+ // submit.
+ if (
+ this.lastSyncNodeType &&
+ this.currentSyncNodeType != this.lastSyncNodeType
+ ) {
+ log.trace(
+ `shouldSubmitForDataChange - nodeType from '${this.lastSyncNodeType}' -> '${this.currentSyncNodeType}'`
+ );
+ return true;
+ }
+ log.trace("shouldSubmitForDataChange - no need to submit");
+ return false;
+ }
+
+ maybeSubmitForDataChange() {
+ if (this._shouldSubmitForDataChange()) {
+ log.info(
+ "Early submission of sync telemetry due to changed IDs/NodeType"
+ );
+ this.finish("idchange"); // this actually submits.
+ this.lastSubmissionTime = Telemetry.msSinceProcessStart();
+ }
+
+ // Only update the last UIDs if we actually know them.
+ let current_uid = fxAccounts.telemetry.getSanitizedUID();
+ if (current_uid) {
+ this.lastUID = current_uid;
+ }
+ if (this.currentSyncNodeType) {
+ this.lastSyncNodeType = this.currentSyncNodeType;
+ }
+ }
+
+ maybeSubmitForInterval() {
+ // We want to submit the ping every `this.submissionInterval` but only when
+ // there's no current sync in progress, otherwise we may end up submitting
+ // the sync and the events caused by it in different pings.
+ if (
+ this.current == null &&
+ Telemetry.msSinceProcessStart() - this.lastSubmissionTime >
+ this.submissionInterval
+ ) {
+ this.finish("schedule");
+ this.lastSubmissionTime = Telemetry.msSinceProcessStart();
+ }
+ }
+
+ onSyncFinished(error) {
+ if (!this.current) {
+ log.warn("onSyncFinished but we aren't recording");
+ return;
+ }
+ this.current.finished(error);
+ this.currentSyncNodeType = this.current.syncNodeType;
+ // We check for "data change" before appending the current sync to payloads,
+ // as it is the current sync which has the data with the new data, and thus
+ // must go in the *next* submission.
+ this.maybeSubmitForDataChange();
+ if (this.payloads.length < this.maxPayloadCount) {
+ this.payloads.push(this.current.toJSON());
+ } else {
+ ++this.discarded;
+ }
+ this.current = null;
+ // If we are submitting due to timing, it's desirable that the most recent
+ // sync is included, so we check after appending `this.current`.
+ this.maybeSubmitForInterval();
+ }
+
+ _addHistogram(hist) {
+ let histogram = Telemetry.getHistogramById(hist);
+ let s = histogram.snapshot();
+ this.histograms[hist] = s;
+ }
+
+ _recordEvent(eventDetails) {
+ this.maybeSubmitForDataChange();
+
+ if (this.events.length >= this.maxEventsCount) {
+ log.warn("discarding event - already queued our maximum", eventDetails);
+ return;
+ }
+
+ let { object, method, value, extra } = eventDetails;
+ if (extra) {
+ extra = normalizeExtraTelemetryFields(extra);
+ eventDetails = { object, method, value, extra };
+ }
+
+ if (!validateTelemetryEvent(eventDetails)) {
+ // we've already logged what the problem is...
+ return;
+ }
+ log.debug("recording event", eventDetails);
+
+ if (extra && Resource.serverTime && !extra.serverTime) {
+ extra.serverTime = String(Resource.serverTime);
+ }
+ let category = "sync";
+ let ts = Math.floor(tryGetMonotonicTimestamp());
+
+ // An event record is a simple array with at least 4 items.
+ let event = [ts, category, method, object];
+ // It may have up to 6 elements if |extra| is defined
+ if (value) {
+ event.push(value);
+ if (extra) {
+ event.push(extra);
+ }
+ } else if (extra) {
+ event.push(null); // a null for the empty value.
+ event.push(extra);
+ }
+ this.events.push(event);
+ this.maybeSubmitForInterval();
+ }
+
+ observe(subject, topic, data) {
+ log.trace(`observed ${topic} ${data}`);
+
+ switch (topic) {
+ case "weave:service:ready":
+ case "weave:service:login:change":
+ case "fxaccounts:new_device_id":
+ this.onAccountInitOrChange();
+ break;
+
+ case "fxaccounts:onlogout":
+ this.onAccountLogout();
+ break;
+
+ /* sync itself state changes */
+ case "weave:service:sync:start":
+ this.onSyncStarted(data);
+ break;
+
+ case "weave:service:sync:finish":
+ if (this._checkCurrent(topic)) {
+ this.onSyncFinished(null);
+ }
+ break;
+
+ case "weave:service:sync:error":
+ // argument needs to be truthy (this should always be the case)
+ this.onSyncFinished(subject || "Unknown");
+ break;
+
+ /* engine sync state changes */
+ case "weave:engine:sync:start":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineStart(data);
+ }
+ break;
+ case "weave:engine:sync:finish":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineStop(data, null);
+ }
+ break;
+
+ case "weave:engine:sync:error":
+ if (this._checkCurrent(topic)) {
+ // argument needs to be truthy (this should always be the case)
+ this.current.onEngineStop(data, subject || "Unknown");
+ }
+ break;
+
+ /* engine counts */
+ case "weave:engine:sync:applied":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineApplied(data, subject);
+ }
+ break;
+
+ case "weave:engine:sync:step":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineStep(data, subject);
+ }
+ break;
+
+ case "weave:engine:sync:uploaded":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineUploaded(data, subject);
+ }
+ break;
+
+ case "weave:engine:validate:finish":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineValidated(data, subject);
+ }
+ break;
+
+ case "weave:engine:validate:error":
+ if (this._checkCurrent(topic)) {
+ this.current.onEngineValidateError(data, subject || "Unknown");
+ }
+ break;
+
+ case "weave:telemetry:event":
+ case "fxa:telemetry:event":
+ this._recordEvent(subject);
+ break;
+
+ case "weave:telemetry:histogram":
+ this._addHistogram(data);
+ break;
+
+ case "weave:telemetry:migration":
+ this._addMigrationRecord(data, subject);
+ break;
+
+ default:
+ log.warn(`unexpected observer topic ${topic}`);
+ break;
+ }
+ }
+
+ // Transform an exception into a standard description. Exposed here for when
+ // this module isn't directly responsible for knowing the transform should
+ // happen (for example, when including an error in the |extra| field of
+ // event telemetry)
+ transformError(error) {
+ // Certain parts of sync will use this pattern as a way to communicate to
+ // processIncoming to abort the processing. However, there's no guarantee
+ // this can only happen then.
+ if (typeof error == "object" && error.code && error.cause) {
+ error = error.cause;
+ }
+ if (Async.isShutdownException(error)) {
+ return { name: "shutdownerror" };
+ }
+
+ if (typeof error === "string") {
+ if (error.startsWith("error.")) {
+ // This is hacky, but I can't imagine that it's not also accurate.
+ return { name: "othererror", error };
+ }
+ error = cleanErrorMessage(error);
+ return { name: "unexpectederror", error };
+ }
+
+ if (error instanceof AuthenticationError) {
+ return { name: "autherror", from: error.source };
+ }
+
+ let httpCode =
+ error.status || (error.response && error.response.status) || error.code;
+
+ if (httpCode) {
+ return { name: "httperror", code: httpCode };
+ }
+
+ if (error.failureCode) {
+ return { name: "othererror", error: error.failureCode };
+ }
+
+ if (error.result) {
+ return { name: "nserror", code: error.result };
+ }
+ // It's probably an Error object, but it also could be some
+ // other object that may or may not override toString to do
+ // something useful.
+ let msg = String(error);
+ if (msg.startsWith("[object")) {
+ // Nothing useful in the default, check for a string "message" property.
+ if (typeof error.message == "string") {
+ msg = String(error.message);
+ } else {
+ // Hopefully it won't come to this...
+ msg = JSON.stringify(error);
+ }
+ }
+ return {
+ name: "unexpectederror",
+ error: cleanErrorMessage(msg),
+ };
+ }
+}
+
+var SyncTelemetry = new SyncTelemetryImpl(ENGINES);