feat(node): Release Health Session Aggregates (#3319)

* feat(node): Request mode sessions

Captures request mode sessions associated with a specific release, and sends them to the Sentry as part of the Release Health functionality.

* Apply suggestions from code review

Co-authored-by: Kamil Ogórek <[email protected]>

* ref: Only have sendSession

* ref: RequestSession getter/setter on scope

* fix: Tests getRequestSession

* ref: initSessionFlusher

* fix: Linter

* fix: CR

* Apply suggestions from code review

Co-authored-by: Kamil Ogórek <[email protected]>

* fix: Lints

Co-authored-by: Daniel Griesser <[email protected]>
Co-authored-by: Kamil Ogórek <[email protected]>

Author: 3 people

packages/core/src/request.ts

@@ -1,4 +1,4 @@
-import { Event, SdkInfo, SentryRequest, Session } from '@sentry/types';
+import { Event, SdkInfo, SentryRequest, SentryRequestType, Session, SessionAggregates } from '@sentry/types';
 
 import { API } from './api';
 
@@ -28,19 +28,21 @@ function enhanceEventWithSdkInfo(event: Event, sdkInfo?: SdkInfo): Event {
 }
 
 /** Creates a SentryRequest from a Session. */
-export function sessionToSentryRequest(session: Session, api: API): SentryRequest {
+export function sessionToSentryRequest(session: Session | SessionAggregates, api: API): SentryRequest {
   const sdkInfo = getSdkMetadataForEnvelopeHeader(api);
   const envelopeHeaders = JSON.stringify({
     sent_at: new Date().toISOString(),
     ...(sdkInfo && { sdk: sdkInfo }),
   });
+  // I know this is hacky but we don't want to add `session` to request type since it's never rate limited
+  const type: SentryRequestType = 'aggregates' in session ? ('sessions' as SentryRequestType) : 'session';
   const itemHeaders = JSON.stringify({
-    type: 'session',
+    type,
   });
 
   return {
     body: `${envelopeHeaders}\n${itemHeaders}\n${JSON.stringify(session)}`,
-    type: 'session',
+    type,
     url: api.getEnvelopeEndpointWithUrlEncodedAuth(),
   };
 }

packages/core/test/lib/request.test.ts

@@ -1,9 +1,19 @@
 import { DebugMeta, Event, SentryRequest, TransactionSamplingMethod } from '@sentry/types';
 
 import { API } from '../../src/api';
-import { eventToSentryRequest } from '../../src/request';
+import { eventToSentryRequest, sessionToSentryRequest } from '../../src/request';
+
+const api = new API('https://[email protected]/12312012', {
+  sdk: {
+    integrations: ['AWSLambda'],
+    name: 'sentry.javascript.browser',
+    version: `12.31.12`,
+    packages: [{ name: 'npm:@sentry/browser', version: `12.31.12` }],
+  },
+});
 
 describe('eventToSentryRequest', () => {
+  let event: Event;
   function parseEnvelopeRequest(request: SentryRequest): any {
     const [envelopeHeaderString, itemHeaderString, eventString] = request.body.split('\n');
 
@@ -14,16 +24,6 @@ describe('eventToSentryRequest', () => {
     };
   }
 
-  const api = new API('https://[email protected]/12312012', {
-    sdk: {
-      integrations: ['AWSLambda'],
-      name: 'sentry.javascript.browser',
-      version: `12.31.12`,
-      packages: [{ name: 'npm:@sentry/browser', version: `12.31.12` }],
-    },
-  });
-  let event: Event;
-
   beforeEach(() => {
     event = {
       contexts: { trace: { trace_id: '1231201211212012', span_id: '12261980', op: 'pageload' } },
@@ -125,3 +125,32 @@ describe('eventToSentryRequest', () => {
     );
   });
 });
+
+describe('sessionToSentryRequest', () => {
+  it('test envelope creation for aggregateSessions', () => {
+    const aggregatedSession = {
+      attrs: { release: '1.0.x', environment: 'prod' },
+      aggregates: [{ started: '2021-04-08T12:18:00.000Z', exited: 2 }],
+    };
+    const result = sessionToSentryRequest(aggregatedSession, api);
+
+    const [envelopeHeaderString, itemHeaderString, sessionString] = result.body.split('\n');
+
+    expect(JSON.parse(envelopeHeaderString)).toEqual(
+      expect.objectContaining({
+        sdk: { name: 'sentry.javascript.browser', version: '12.31.12' },
+      }),
+    );
+    expect(JSON.parse(itemHeaderString)).toEqual(
+      expect.objectContaining({
+        type: 'sessions',
+      }),
+    );
+    expect(JSON.parse(sessionString)).toEqual(
+      expect.objectContaining({
+        attrs: { release: '1.0.x', environment: 'prod' },
+        aggregates: [{ started: '2021-04-08T12:18:00.000Z', exited: 2 }],
+      }),
+    );
+  });
+});

packages/hub/src/index.ts

@@ -1,7 +1,7 @@
 // eslint-disable-next-line deprecation/deprecation
 export { Carrier, DomainAsCarrier, Layer } from './interfaces';
 export { addGlobalEventProcessor, Scope } from './scope';
-export { Session } from './session';
+export { Session, SessionFlusher } from './session';
 export {
   // eslint-disable-next-line deprecation/deprecation
   getActiveDomain,

packages/hub/src/scope.ts

@@ -10,6 +10,7 @@ import {
   Extra,
   Extras,
   Primitive,
+  RequestSession,
   Scope as ScopeInterface,
   ScopeContext,
   Severity,
@@ -65,6 +66,9 @@ export class Scope implements ScopeInterface {
   /** Session */
   protected _session?: Session;
 
+  /** Request Mode Session Status */
+  protected _requestSession?: RequestSession;
+
   /**
    * Inherit values from the parent scope.
    * @param scope to clone.
@@ -83,6 +87,7 @@ export class Scope implements ScopeInterface {
       newScope._transactionName = scope._transactionName;
       newScope._fingerprint = scope._fingerprint;
       newScope._eventProcessors = [...scope._eventProcessors];
+      newScope._requestSession = scope._requestSession;
     }
     return newScope;
   }
@@ -122,6 +127,21 @@ export class Scope implements ScopeInterface {
     return this._user;
   }
 
+  /**
+   * @inheritDoc
+   */
+  public getRequestSession(): RequestSession | undefined {
+    return this._requestSession;
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public setRequestSession(requestSession?: RequestSession): this {
+    this._requestSession = requestSession;
+    return this;
+  }
+
   /**
    * @inheritDoc
    */
@@ -297,6 +317,9 @@ export class Scope implements ScopeInterface {
       if (captureContext._fingerprint) {
         this._fingerprint = captureContext._fingerprint;
       }
+      if (captureContext._requestSession) {
+        this._requestSession = captureContext._requestSession;
+      }
     } else if (isPlainObject(captureContext)) {
       // eslint-disable-next-line no-param-reassign
       captureContext = captureContext as ScopeContext;
@@ -312,6 +335,9 @@ export class Scope implements ScopeInterface {
       if (captureContext.fingerprint) {
         this._fingerprint = captureContext.fingerprint;
       }
+      if (captureContext.requestSession) {
+        this._requestSession = captureContext.requestSession;
+      }
     }
 
     return this;
@@ -329,6 +355,7 @@ export class Scope implements ScopeInterface {
     this._level = undefined;
     this._transactionName = undefined;
     this._fingerprint = undefined;
+    this._requestSession = undefined;
     this._span = undefined;
     this._session = undefined;
     this._notifyScopeListeners();

packages/hub/src/session.ts

@@ -1,5 +1,16 @@
-import { Session as SessionInterface, SessionContext, SessionStatus } from '@sentry/types';
-import { dropUndefinedKeys, uuid4 } from '@sentry/utils';
+import {
+  AggregationCounts,
+  RequestSessionStatus,
+  Session as SessionInterface,
+  SessionAggregates,
+  SessionContext,
+  SessionFlusherLike,
+  SessionStatus,
+  Transport,
+} from '@sentry/types';
+import { dropUndefinedKeys, logger, uuid4 } from '@sentry/utils';
+
+import { getCurrentHub } from './hub';
 
 /**
  * @inheritdoc
@@ -123,3 +134,119 @@ export class Session implements SessionInterface {
     });
   }
 }
+
+type ReleaseHealthAttributes = {
+  environment?: string;
+  release: string;
+};
+
+/**
+ * @inheritdoc
+ */
+export class SessionFlusher implements SessionFlusherLike {
+  public readonly flushTimeout: number = 60;
+  private _pendingAggregates: Record<number, AggregationCounts> = {};
+  private _sessionAttrs: ReleaseHealthAttributes;
+  private _intervalId: ReturnType<typeof setInterval>;
+  private _isEnabled: boolean = true;
+  private _transport: Transport;
+
+  constructor(transport: Transport, attrs: ReleaseHealthAttributes) {
+    this._transport = transport;
+    // Call to setInterval, so that flush is called every 60 seconds
+    this._intervalId = setInterval(() => this.flush(), this.flushTimeout * 1000);
+    this._sessionAttrs = attrs;
+  }
+
+  /** Sends session aggregates to Transport */
+  public sendSessionAggregates(sessionAggregates: SessionAggregates): void {
+    if (!this._transport.sendSession) {
+      logger.warn("Dropping session because custom transport doesn't implement sendSession");
+      return;
+    }
+    this._transport.sendSession(sessionAggregates).then(null, reason => {
+      logger.error(`Error while sending session: ${reason}`);
+    });
+  }
+
+  /** Checks if `pendingAggregates` has entries, and if it does flushes them by calling `sendSessions` */
+  public flush(): void {
+    const sessionAggregates = this.getSessionAggregates();
+    if (sessionAggregates.aggregates.length === 0) {
+      return;
+    }
+    this._pendingAggregates = {};
+    this.sendSessionAggregates(sessionAggregates);
+  }
+
+  /** Massages the entries in `pendingAggregates` and returns aggregated sessions */
+  public getSessionAggregates(): SessionAggregates {
+    const aggregates: AggregationCounts[] = Object.keys(this._pendingAggregates).map((key: string) => {
+      return this._pendingAggregates[parseInt(key)];
+    });
+
+    const sessionAggregates: SessionAggregates = {
+      attrs: this._sessionAttrs,
+      aggregates,
+    };
+    return dropUndefinedKeys(sessionAggregates);
+  }
+
+  /** JSDoc */
+  public close(): void {
+    clearInterval(this._intervalId);
+    this._isEnabled = false;
+    this.flush();
+  }
+
+  /**
+   * Wrapper function for _incrementSessionStatusCount that checks if the instance of SessionFlusher is enabled then
+   * fetches the session status of the request from `Scope.getRequestSession().status` on the scope and passes them to
+   * `_incrementSessionStatusCount` along with the start date
+   */
+  public incrementSessionStatusCount(): void {
+    if (!this._isEnabled) {
+      return;
+    }
+    const scope = getCurrentHub().getScope();
+    const requestSession = scope?.getRequestSession();
+
+    if (requestSession && requestSession.status) {
+      this._incrementSessionStatusCount(requestSession.status, new Date());
+      // This is not entirely necessarily but is added as a safe guard to indicate the bounds of a request and so in
+      // case captureRequestSession is called more than once to prevent double count
+      scope?.setRequestSession(undefined);
+
+      /* eslint-enable @typescript-eslint/no-unsafe-member-access */
+    }
+  }
+
+  /**
+   * Increments status bucket in pendingAggregates buffer (internal state) corresponding to status of
+   * the session received
+   */
+  private _incrementSessionStatusCount(status: RequestSessionStatus, date: Date): number {
+    // Truncate minutes and seconds on Session Started attribute to have one minute bucket keys
+    const sessionStartedTrunc = new Date(date).setSeconds(0, 0);
+    this._pendingAggregates[sessionStartedTrunc] = this._pendingAggregates[sessionStartedTrunc] || {};
+
+    // corresponds to aggregated sessions in one specific minute bucket
+    // for example, {"started":"2021-03-16T08:00:00.000Z","exited":4, "errored": 1}
+    const aggregationCounts: AggregationCounts = this._pendingAggregates[sessionStartedTrunc];
+    if (!aggregationCounts.started) {
+      aggregationCounts.started = new Date(sessionStartedTrunc).toISOString();
+    }
+
+    switch (status) {
+      case RequestSessionStatus.Errored:
+        aggregationCounts.errored = (aggregationCounts.errored || 0) + 1;
+        return aggregationCounts.errored;
+      case RequestSessionStatus.Ok:
+        aggregationCounts.exited = (aggregationCounts.exited || 0) + 1;
+        return aggregationCounts.exited;
+      case RequestSessionStatus.Crashed:
+        aggregationCounts.crashed = (aggregationCounts.crashed || 0) + 1;
+        return aggregationCounts.crashed;
+    }
+  }
+}