diff --git a/package-lock.json b/package-lock.json
index 646424c70fa..4dbefd14af0 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -5372,6 +5372,12 @@
         "@types/node": "*"
       }
     },
+    "@types/uuid": {
+      "version": "7.0.3",
+      "resolved": "https://registry.npmjs.org/@types/uuid/-/uuid-7.0.3.tgz",
+      "integrity": "sha512-PUdqTZVrNYTNcIhLHkiaYzoOIaUi5LFg/XLerAdgvwQrUCx+oSbtoBze1AMyvYbcwzUSNC+Isl58SM4Sm/6COw==",
+      "dev": true
+    },
     "@types/ws": {
       "version": "7.2.4",
       "resolved": "https://registry.npmjs.org/@types/ws/-/ws-7.2.4.tgz",
@@ -5588,7 +5594,15 @@
         "apollo-server-errors": "file:packages/apollo-server-errors",
         "apollo-server-plugin-base": "file:packages/apollo-server-plugin-base",
         "apollo-server-types": "file:packages/apollo-server-types",
-        "async-retry": "^1.2.1"
+        "async-retry": "^1.2.1",
+        "uuid": "^8.0.0"
+      },
+      "dependencies": {
+        "uuid": {
+          "version": "8.0.0",
+          "resolved": "https://registry.npmjs.org/uuid/-/uuid-8.0.0.tgz",
+          "integrity": "sha512-jOXGuXZAWdsTH7eZLtyXMqUb9EcWMGZNbL9YcGBJl4MH4nrxHmZJhEHvyLFrkxo+28uLb/NYRcStH48fnD0Vzw=="
+        }
       }
     },
     "apollo-engine-reporting-protobuf": {
diff --git a/package.json b/package.json
index 3454ef8a065..30b9b00e3b8 100644
--- a/package.json
+++ b/package.json
@@ -86,6 +86,7 @@
     "@types/supertest": "^2.0.8",
     "@types/test-listen": "1.1.0",
     "@types/type-is": "1.6.3",
+    "@types/uuid": "^7.0.3",
     "@types/ws": "7.2.4",
     "apollo-fetch": "0.7.0",
     "apollo-link": "1.2.14",
diff --git a/packages/apollo-engine-reporting-protobuf/src/reports.proto b/packages/apollo-engine-reporting-protobuf/src/reports.proto
index 9f106f1c50b..0c6bc319d78 100644
--- a/packages/apollo-engine-reporting-protobuf/src/reports.proto
+++ b/packages/apollo-engine-reporting-protobuf/src/reports.proto
@@ -269,8 +269,10 @@ message ReportHeader {
 	string uname = 9;
 	// eg "current", "prod"
 	string schema_tag = 10;
-	// The hex representation of the sha512 of the introspection response
-	string schema_hash = 11;
+	// An id that is used to represent the schema to Apollo Graph Manager
+	// Using this in place of what used to be schema_hash, since that is no longer
+	// attached to a schema in the backend.
+	string executable_schema_id = 11;
 
 	reserved 3; // removed string service = 3;
 }
diff --git a/packages/apollo-engine-reporting/package.json b/packages/apollo-engine-reporting/package.json
index 93259c7143f..9ce287bb57f 100644
--- a/packages/apollo-engine-reporting/package.json
+++ b/packages/apollo-engine-reporting/package.json
@@ -18,7 +18,8 @@
     "apollo-server-errors": "file:../apollo-server-errors",
     "apollo-server-plugin-base": "file:../apollo-server-plugin-base",
     "apollo-server-types": "file:../apollo-server-types",
-    "async-retry": "^1.2.1"
+    "async-retry": "^1.2.1",
+    "uuid": "^8.0.0"
   },
   "peerDependencies": {
     "graphql": "^0.12.0 || ^0.13.0 || ^14.0.0 || ^15.0.0"
diff --git a/packages/apollo-engine-reporting/src/__tests__/agent.test.ts b/packages/apollo-engine-reporting/src/__tests__/agent.test.ts
index ff6ac43ab6a..565cc5e50d5 100644
--- a/packages/apollo-engine-reporting/src/__tests__/agent.test.ts
+++ b/packages/apollo-engine-reporting/src/__tests__/agent.test.ts
@@ -2,7 +2,9 @@ import {
   signatureCacheKey,
   handleLegacyOptions,
   EngineReportingOptions,
-} from '../agent';
+  computeExecutableSchemaId
+} from "../agent";
+import { buildSchema } from "graphql";
 
 describe('signature cache key', () => {
   it('generates without the operationName', () => {
@@ -16,6 +18,59 @@ describe('signature cache key', () => {
   });
 });
 
+describe('Executable Schema Id', () => {
+  const unsortedGQLSchemaDocument = `
+      directive @example on FIELD
+      union AccountOrUser = Account | User
+      type Query {
+        userOrAccount(name: String, id: String): AccountOrUser
+      }
+
+      type User {
+        accounts: [Account!]
+        email: String
+        name: String!
+      }
+
+      type Account {
+        name: String!
+        id: ID!
+      }
+    `;
+
+  const sortedGQLSchemaDocument = `
+      directive @example on FIELD
+      union AccountOrUser = Account | User
+
+      type Account {
+        name: String!
+        id: ID!
+      }
+
+      type Query {
+        userOrAccount(id: String, name: String): AccountOrUser
+      }
+
+      type User {
+        accounts: [Account!]
+        email: String
+        name: String!
+      }
+
+    `;
+  it('does not normalize GraphQL schemas', () => {
+    expect(computeExecutableSchemaId(buildSchema(unsortedGQLSchemaDocument))).not.toEqual(
+      computeExecutableSchemaId(buildSchema(sortedGQLSchemaDocument))
+    );
+  });
+  it('does not normalize strings', () => {
+    expect(computeExecutableSchemaId(unsortedGQLSchemaDocument)).not.toEqual(
+      computeExecutableSchemaId(sortedGQLSchemaDocument)
+    );
+  });
+});
+
+
 describe("test handleLegacyOptions(), which converts the deprecated privateVariable and privateHeaders options to the new options' formats", () => {
   it('Case 1: privateVariables/privateHeaders == False; same as all', () => {
     const optionsPrivateFalse: EngineReportingOptions<any> = {
diff --git a/packages/apollo-engine-reporting/src/__tests__/plugin.test.ts b/packages/apollo-engine-reporting/src/__tests__/plugin.test.ts
index 2a8d3313b12..56c2809a3db 100644
--- a/packages/apollo-engine-reporting/src/__tests__/plugin.test.ts
+++ b/packages/apollo-engine-reporting/src/__tests__/plugin.test.ts
@@ -1,14 +1,13 @@
 import { makeExecutableSchema, addMockFunctionsToSchema } from 'graphql-tools';
-import { graphql, GraphQLError } from 'graphql';
+import { graphql, GraphQLError, printSchema } from 'graphql';
 import { Request } from 'node-fetch';
 import { makeTraceDetails, makeHTTPRequestHeaders, plugin } from '../plugin';
 import { Headers } from 'apollo-server-env';
-import { AddTraceArgs } from '../agent';
+import { AddTraceArgs, computeExecutableSchemaId } from '../agent';
 import { Trace } from 'apollo-engine-reporting-protobuf';
 import pluginTestHarness from 'apollo-server-core/dist/utils/pluginTestHarness';
 
-it('trace construction', async () => {
-  const typeDefs = `
+const typeDefs = `
   type User {
     id: Int
     name: String
@@ -31,7 +30,7 @@ it('trace construction', async () => {
   }
 `;
 
-  const query = `
+const query = `
     query q {
       author(id: 5) {
         name
@@ -43,6 +42,154 @@ it('trace construction', async () => {
     }
 `;
 
+describe('schema reporting', () => {
+  const schema = makeExecutableSchema({ typeDefs });
+  addMockFunctionsToSchema({ schema });
+
+  const addTrace = jest.fn();
+  const startSchemaReporting = jest.fn();
+  const executableSchemaIdGenerator = jest.fn(computeExecutableSchemaId);
+
+  beforeEach(() => {
+    addTrace.mockClear();
+    startSchemaReporting.mockClear();
+    executableSchemaIdGenerator.mockClear();
+  });
+
+  it('starts reporting if enabled', async () => {
+    const pluginInstance = plugin(
+      {
+        experimental_schemaReporting: true,
+      },
+      addTrace,
+      {
+        startSchemaReporting,
+        executableSchemaIdGenerator,
+      },
+    );
+
+    await pluginTestHarness({
+      pluginInstance,
+      schema,
+      graphqlRequest: {
+        query,
+        operationName: 'q',
+        extensions: {
+          clientName: 'testing suite',
+        },
+        http: new Request('http://localhost:123/foo'),
+      },
+      executor: async ({ request: { query: source } }) => {
+        return await graphql({
+          schema,
+          source,
+        });
+      },
+    });
+
+    expect(startSchemaReporting).toBeCalledTimes(1);
+    expect(startSchemaReporting).toBeCalledWith({
+      executableSchema: printSchema(schema),
+      executableSchemaId: executableSchemaIdGenerator(schema),
+    });
+  });
+
+  it('uses the override schema', async () => {
+    const pluginInstance = plugin(
+      {
+        experimental_schemaReporting: true,
+        experimental_overrideReportedSchema: typeDefs,
+      },
+      addTrace,
+      {
+        startSchemaReporting,
+        executableSchemaIdGenerator,
+      },
+    );
+
+    await pluginTestHarness({
+      pluginInstance,
+      schema,
+      graphqlRequest: {
+        query,
+        operationName: 'q',
+        extensions: {
+          clientName: 'testing suite',
+        },
+        http: new Request('http://localhost:123/foo'),
+      },
+      executor: async ({ request: { query: source } }) => {
+        return await graphql({
+          schema,
+          source,
+        });
+      },
+    });
+
+    const expectedExecutableSchemaId = executableSchemaIdGenerator(typeDefs);
+    expect(startSchemaReporting).toBeCalledTimes(1);
+    expect(startSchemaReporting).toBeCalledWith({
+      executableSchema: typeDefs,
+      executableSchemaId: expectedExecutableSchemaId,
+    });
+
+    // Get the first argument from the first time this is called.
+    // Not using called with because that has to be exhaustive and this isn't
+    // testing trace generation
+    expect(addTrace).toBeCalledWith(
+      expect.objectContaining({
+        executableSchemaId: expectedExecutableSchemaId,
+      }),
+    );
+  });
+
+  it('uses the same executable schema id for metric reporting', async () => {
+    const pluginInstance = plugin(
+      {
+        experimental_schemaReporting: true,
+      },
+      addTrace,
+      {
+        startSchemaReporting,
+        executableSchemaIdGenerator,
+      },
+    );
+
+    await pluginTestHarness({
+      pluginInstance,
+      schema,
+      graphqlRequest: {
+        query,
+        operationName: 'q',
+        extensions: {
+          clientName: 'testing suite',
+        },
+        http: new Request('http://localhost:123/foo'),
+      },
+      executor: async ({ request: { query: source } }) => {
+        return await graphql({
+          schema,
+          source,
+        });
+      },
+    });
+
+    const expectedExecutableSchemaId = executableSchemaIdGenerator(schema);
+    expect(startSchemaReporting).toBeCalledTimes(1);
+    expect(startSchemaReporting).toBeCalledWith({
+      executableSchema: printSchema(schema),
+      executableSchemaId: expectedExecutableSchemaId,
+    });
+    // Get the first argument from the first time this is called.
+    // Not using called with because that has to be exhaustive and this isn't
+    // testing trace generation
+    expect(addTrace.mock.calls[0][0].executableSchemaId).toBe(
+      expectedExecutableSchemaId,
+    );
+  });
+});
+
+it('trace construction', async () => {
   const schema = makeExecutableSchema({ typeDefs });
   addMockFunctionsToSchema({ schema });
 
@@ -50,10 +197,21 @@ it('trace construction', async () => {
   async function addTrace(args: AddTraceArgs) {
     traces.push(args);
   }
+  const startSchemaReporting = jest.fn();
+  const executableSchemaIdGenerator = jest.fn();
 
-  const pluginInstance = plugin({ /* no options!*/ }, addTrace);
+  const pluginInstance = plugin(
+    {
+      /* no options!*/
+    },
+    addTrace,
+    {
+      startSchemaReporting,
+      executableSchemaIdGenerator,
+    },
+  );
 
-  pluginTestHarness({
+  await pluginTestHarness({
     pluginInstance,
     schema,
     graphqlRequest: {
@@ -64,7 +222,7 @@ it('trace construction', async () => {
       },
       http: new Request('http://localhost:123/foo'),
     },
-    executor: async ({ request: { query: source }}) => {
+    executor: async ({ request: { query: source } }) => {
       return await graphql({
         schema,
         source,
@@ -260,7 +418,7 @@ describe('variableJson output for sendVariableValues transform: custom function
     ).toEqual(JSON.stringify(null));
   });
 
-  const errorThrowingModifier = (input: {
+  const errorThrowingModifier = (_input: {
     variables: Record<string, any>;
   }): Record<string, any> => {
     throw new GraphQLError('testing error handling');
diff --git a/packages/apollo-engine-reporting/src/__tests__/schemaReporter.test.ts b/packages/apollo-engine-reporting/src/__tests__/schemaReporter.test.ts
new file mode 100644
index 00000000000..44b317c7fd0
--- /dev/null
+++ b/packages/apollo-engine-reporting/src/__tests__/schemaReporter.test.ts
@@ -0,0 +1,173 @@
+import nock from 'nock';
+import { reportServerInfoGql, SchemaReporter } from '../schemaReporter';
+
+function mockReporterRequest(url: any, variables?: any) {
+  if (variables)
+    return nock(url).post(
+      '/',
+      JSON.stringify({
+        query: reportServerInfoGql,
+        operationName: 'ReportServerInfo',
+        variables,
+      }),
+    );
+  return nock(url).post('/');
+}
+
+beforeEach(() => {
+  if (!nock.isActive()) nock.activate();
+});
+
+afterEach(() => {
+  expect(nock.isDone()).toBeTruthy();
+  nock.cleanAll();
+  nock.restore();
+});
+
+const serverInfo = {
+  bootId: 'string',
+  executableSchemaId: 'string',
+  graphVariant: 'string',
+};
+
+const url = 'http://localhost:4000';
+
+describe('Schema reporter', () => {
+  it('return correct values if no errors', async () => {
+    const schemaReporter = new SchemaReporter(
+      serverInfo,
+      'schemaSdl',
+      'apiKey',
+      url,
+    );
+    mockReporterRequest(url).reply(200, {
+      data: {
+        me: {
+          __typename: 'ServiceMutation',
+          reportServerInfo: {
+            __typename: 'ReportServerInfoResponse',
+            inSeconds: 30,
+            withExecutableSchema: false,
+          },
+        },
+      },
+    });
+
+    let {
+      inSeconds,
+      withExecutableSchema,
+    } = await schemaReporter.reportServerInfo(false);
+    expect(inSeconds).toBe(30);
+    expect(withExecutableSchema).toBe(false);
+
+    mockReporterRequest(url).reply(200, {
+      data: {
+        me: {
+          __typename: 'ServiceMutation',
+          reportServerInfo: {
+            __typename: 'ReportServerInfoResponse',
+            inSeconds: 60,
+            withExecutableSchema: true,
+          },
+        },
+      },
+    });
+    ({
+      inSeconds,
+      withExecutableSchema,
+    } = await schemaReporter.reportServerInfo(false));
+    expect(inSeconds).toBe(60);
+    expect(withExecutableSchema).toBe(true);
+  });
+
+  it('throws on 500 response', async () => {
+    const schemaReporter = new SchemaReporter(
+      serverInfo,
+      'schemaSdl',
+      'apiKey',
+      url,
+    );
+    mockReporterRequest(url).reply(500, {
+      data: {
+        me: {
+          reportServerInfo: {
+            __typename: 'ReportServerInfoResponse',
+            inSeconds: 30,
+            withExecutableSchema: false,
+          },
+        },
+      },
+    });
+
+    await expect(
+      schemaReporter.reportServerInfo(false),
+    ).rejects.toThrowErrorMatchingInlineSnapshot(
+      `"An unexpected HTTP status code (500) was encountered during schema reporting."`,
+    );
+  });
+
+  it('throws on 200 malformed response', async () => {
+    const schemaReporter = new SchemaReporter(
+      serverInfo,
+      'schemaSdl',
+      'apiKey',
+      url,
+    );
+    mockReporterRequest(url).reply(200, {
+      data: {
+        me: {
+          reportServerInfo: {
+            __typename: 'ReportServerInfoResponse',
+          },
+        },
+      },
+    });
+
+    await expect(
+      schemaReporter.reportServerInfo(false),
+    ).rejects.toThrowErrorMatchingInlineSnapshot(
+      `"Unexpected response shape from Apollo Graph Manager when reporting server information for schema reporting. If this continues, please reach out to support@apollographql.com. Received response: {\\"me\\":{\\"reportServerInfo\\":{\\"__typename\\":\\"ReportServerInfoResponse\\"}}}"`,
+    );
+
+    mockReporterRequest(url).reply(200, {
+      data: {
+        me: {
+          __typename: 'UserMutation',
+        },
+      },
+    });
+    await expect(
+      schemaReporter.reportServerInfo(false),
+    ).rejects.toThrowErrorMatchingInlineSnapshot(
+      `"This server was configured with an API key for a user. Only a service's API key may be used for schema reporting. Please visit the settings for this graph at https://engine.apollographql.com/ to obtain an API key for a service."`,
+    );
+  });
+
+  it('sends schema if withExecutableSchema is true.', async () => {
+    const schemaReporter = new SchemaReporter(
+      serverInfo,
+      'schemaSdl',
+      'apiKey',
+      url,
+    );
+
+    const variables = {
+      info: serverInfo,
+      executableSchema: 'schemaSdl'
+    };
+    mockReporterRequest(url, variables).reply(200, {
+      data: {
+        me: {
+          __typename: 'ServiceMutation',
+          reportServerInfo: {
+            __typename: 'ReportServerInfoResponse',
+            inSeconds: 30,
+            withExecutableSchema: false,
+          },
+        },
+      },
+    });
+
+    await schemaReporter.reportServerInfo(true);
+  });
+});
diff --git a/packages/apollo-engine-reporting/src/agent.ts b/packages/apollo-engine-reporting/src/agent.ts
index c7e6a28d083..bd3a13b327f 100644
--- a/packages/apollo-engine-reporting/src/agent.ts
+++ b/packages/apollo-engine-reporting/src/agent.ts
@@ -1,21 +1,29 @@
 import os from 'os';
 import { gzip } from 'zlib';
-import { DocumentNode, GraphQLError } from 'graphql';
+import {
+  DocumentNode,
+  GraphQLError,
+  GraphQLSchema,
+  printSchema,
+} from 'graphql';
 import {
   ReportHeader,
   Trace,
   Report,
-  TracesAndStats
+  TracesAndStats,
 } from 'apollo-engine-reporting-protobuf';
 
 import { fetch, RequestAgent, Response } from 'apollo-server-env';
 import retry from 'async-retry';
 
 import { plugin } from './plugin';
-import { GraphQLRequestContext, Logger, SchemaHash } from 'apollo-server-types';
+import { GraphQLRequestContext, Logger } from 'apollo-server-types';
 import { InMemoryLRUCache } from 'apollo-server-caching';
 import { defaultEngineReportingSignature } from 'apollo-graphql';
-import { ApolloServerPlugin } from "apollo-server-plugin-base";
+import { ApolloServerPlugin } from 'apollo-server-plugin-base';
+import { reportingLoop, SchemaReporter } from './schemaReporter';
+import { v4 as uuidv4 } from 'uuid';
+import { createHash } from 'crypto';
 
 let warnedOnDeprecatedApiKey = false;
 
@@ -120,9 +128,15 @@ export interface EngineReportingOptions<TContext> {
    */
   maxUncompressedReportSize?: number;
   /**
+   * [DEPRECATED] this option was replaced by tracesEndpointUrl
    * The URL of the Engine report ingress server.
    */
   endpointUrl?: string;
+  /**
+   * The URL to the Apollo Graph Manager ingress endpoint.
+   * (Previously, this was `endpointUrl`, which will be removed in AS3).
+   */
+  tracesEndpointUrl?: string;
   /**
    * If set, prints all reports as JSON when they are sent.
    */
@@ -239,6 +253,47 @@ export interface EngineReportingOptions<TContext> {
    */
   generateClientInfo?: GenerateClientInfo<TContext>;
 
+  /**
+   * **(Experimental)** Enable schema reporting from this server with
+   * Apollo Graph Manager.
+   *
+   * The use of this option avoids the need to rgister schemas manually within
+   * CI deployment pipelines using `apollo schema:push` by periodically
+   * reporting this server's schema (when changes are detected) along with
+   * additional details about its runtime environment to Apollo Graph Manager.
+   *
+   * See [our _preview
+   * documentation_](https://github.com/apollographql/apollo-schema-reporting-preview-docs)
+   * for more information.
+   */
+  experimental_schemaReporting?: boolean;
+
+  /**
+   * Override the reported schema that is reported to AGM.
+   * This schema does not go through any normalizations and the string is directly sent to Apollo Graph Manager.
+   * This would be useful for comments or other ordering and whitespace changes that get stripped when generating a `GraphQLSchema`
+   */
+  experimental_overrideReportedSchema?: string;
+
+  /**
+   * The schema reporter waits before starting reporting.
+   * By default, the report waits some random amount of time between 0 and 10 seconds.
+   * A longer interval leads to more staggered starts which means it is less likely
+   * multiple servers will get asked to upload the same schema.
+   *
+   * If this server runs in lambda or in other constrained environments it would be useful
+   * to decrease the schema reporting max wait time to be less than default.
+   *
+   * This number will be the max for the range in ms that the schema reporter will
+   * wait before starting to report.
+   */
+  experimental_schemaReportingInitialDelayMaxMs?: number;
+
+  /**
+   * The URL to use for reporting schemas.
+   */
+  schemaReportingUrl?: string;
+
   /**
    * A logger interface to be used for output and errors.  When not provided
    * it will default to the server's own `logger` implementation and use
@@ -251,7 +306,7 @@ export interface AddTraceArgs {
   trace: Trace;
   operationName: string;
   queryHash: string;
-  schemaHash: SchemaHash;
+  executableSchemaId: string;
   source?: string;
   document?: DocumentNode;
 }
@@ -270,27 +325,42 @@ const serviceHeaderDefaults = {
 export class EngineReportingAgent<TContext = any> {
   private readonly options: EngineReportingOptions<TContext>;
   private readonly apiKey: string;
-  private logger: Logger = console;
-  private graphVariant: string;
-  private reports: { [schemaHash: string]: Report } = Object.create(
+  private readonly logger: Logger = console;
+  private readonly graphVariant: string;
+
+  private reports: { [executableSchemaId: string]: Report } = Object.create(
+    null,
+  );
+  private reportSizes: { [executableSchemaId: string]: number } = Object.create(
     null,
   );
-  private reportSizes: { [schemaHash: string]: number } = Object.create(null);
+
   private reportTimer: any; // timer typing is weird and node-specific
   private readonly sendReportsImmediately?: boolean;
   private stopped: boolean = false;
-  private reportHeaders: { [schemaHash: string]: ReportHeader } = Object.create(
-    null,
-  );
+  private reportHeaders: {
+    [executableSchemaId: string]: ReportHeader;
+  } = Object.create(null);
   private signatureCache: InMemoryLRUCache<string>;
 
   private signalHandlers = new Map<NodeJS.Signals, NodeJS.SignalsListener>();
 
+  private currentSchemaReporter?: SchemaReporter;
+  private readonly bootId: string;
+  private lastSeenExecutableSchemaToId?: {
+    executableSchema: string | GraphQLSchema;
+    executableSchemaId: string;
+  };
+
+  private readonly tracesEndpointUrl: string;
+
   public constructor(options: EngineReportingOptions<TContext> = {}) {
     this.options = options;
     this.apiKey = getEngineApiKey({engine: this.options, skipWarn: false, logger: this.logger});
     if (options.logger) this.logger = options.logger;
+    this.bootId = uuidv4();
     this.graphVariant = getEngineGraphVariant(options, this.logger) || '';
+
     if (!this.apiKey) {
       throw new Error(
         `To use EngineReportingAgent, you must specify an API key via the apiKey option or the APOLLO_KEY environment variable.`,
@@ -325,12 +395,41 @@ export class EngineReportingAgent<TContext = any> {
       });
     }
 
+    if (this.options.endpointUrl) {
+      this.logger.warn(
+        '[deprecated] The `endpointUrl` option within `engine` has been renamed to `tracesEndpointUrl`.',
+      );
+    }
+    this.tracesEndpointUrl =
+      (this.options.endpointUrl ||
+        this.options.tracesEndpointUrl ||
+        'https://engine-report.apollodata.com') + '/api/ingress/traces';
+
     // Handle the legacy options: privateVariables and privateHeaders
     handleLegacyOptions(this.options);
   }
 
+  private executableSchemaIdGenerator(schema: string | GraphQLSchema) {
+    if (this.lastSeenExecutableSchemaToId?.executableSchema === schema) {
+      return this.lastSeenExecutableSchemaToId.executableSchemaId;
+    }
+    const id = computeExecutableSchemaId(schema);
+
+    // We override this variable every time we get a new schema so we cache
+    // the last seen value. It mostly a cached pair.
+    this.lastSeenExecutableSchemaToId = {
+      executableSchema: schema,
+      executableSchemaId: id,
+    };
+
+    return id;
+  }
+
   public newPlugin(): ApolloServerPlugin<TContext> {
-    return plugin(this.options, this.addTrace.bind(this));
+    return plugin(this.options, this.addTrace.bind(this), {
+      startSchemaReporting: this.startSchemaReporting.bind(this),
+      executableSchemaIdGenerator: this.executableSchemaIdGenerator.bind(this),
+    });
   }
 
   public async addTrace({
@@ -339,23 +438,23 @@ export class EngineReportingAgent<TContext = any> {
     document,
     operationName,
     source,
-    schemaHash,
+    executableSchemaId,
   }: AddTraceArgs): Promise<void> {
     // Ignore traces that come in after stop().
     if (this.stopped) {
       return;
     }
 
-    if (!(schemaHash in this.reports)) {
-      this.reportHeaders[schemaHash] = new ReportHeader({
+    if (!(executableSchemaId in this.reports)) {
+      this.reportHeaders[executableSchemaId] = new ReportHeader({
         ...serviceHeaderDefaults,
-        schemaHash,
+        executableSchemaId: executableSchemaId,
         schemaTag: this.graphVariant,
       });
       // initializes this.reports[reportHash]
-      this.resetReport(schemaHash);
+      this.resetReport(executableSchemaId);
     }
-    const report = this.reports[schemaHash];
+    const report = this.reports[executableSchemaId];
 
     const protobufError = Trace.verify(trace);
     if (protobufError) {
@@ -380,28 +479,26 @@ export class EngineReportingAgent<TContext = any> {
     (report.tracesPerQuery[statsReportKey] as any).encodedTraces.push(
       encodedTrace,
     );
-    this.reportSizes[schemaHash] +=
+    this.reportSizes[executableSchemaId] +=
       encodedTrace.length + Buffer.byteLength(statsReportKey);
 
     // If the buffer gets big (according to our estimate), send.
     if (
       this.sendReportsImmediately ||
-      this.reportSizes[schemaHash] >=
+      this.reportSizes[executableSchemaId] >=
         (this.options.maxUncompressedReportSize || 4 * 1024 * 1024)
     ) {
-      await this.sendReportAndReportErrors(schemaHash);
+      await this.sendReportAndReportErrors(executableSchemaId);
     }
   }
 
   public async sendAllReports(): Promise<void> {
-    await Promise.all(
-      Object.keys(this.reports).map(hash => this.sendReport(hash)),
-    );
+    await Promise.all(Object.keys(this.reports).map(id => this.sendReport(id)));
   }
 
-  public async sendReport(schemaHash: string): Promise<void> {
-    const report = this.reports[schemaHash];
-    this.resetReport(schemaHash);
+  public async sendReport(executableSchemaId: string): Promise<void> {
+    const report = this.reports[executableSchemaId];
+    this.resetReport(executableSchemaId);
 
     if (Object.keys(report.tracesPerQuery).length === 0) {
       return;
@@ -449,16 +546,12 @@ export class EngineReportingAgent<TContext = any> {
       });
     });
 
-    const endpointUrl =
-      (this.options.endpointUrl || 'https://engine-report.apollodata.com') +
-      '/api/ingress/traces';
-
     // Wrap fetch with async-retry for automatic retrying
     const response: Response = await retry(
       // Retry on network errors and 5xx HTTP
       // responses.
       async () => {
-        const curResponse = await fetch(endpointUrl, {
+        const curResponse = await fetch(this.tracesEndpointUrl, {
           method: 'POST',
           headers: {
             'user-agent': 'apollo-engine-reporting',
@@ -511,6 +604,59 @@ export class EngineReportingAgent<TContext = any> {
     }
   }
 
+  public startSchemaReporting({
+    executableSchemaId,
+    executableSchema,
+  }: {
+    executableSchemaId: string;
+    executableSchema: string;
+  }) {
+    if (this.currentSchemaReporter) {
+      this.currentSchemaReporter.stop();
+    }
+
+    const serverInfo = {
+      bootId: this.bootId,
+      graphVariant: this.graphVariant,
+      // The infra environment in which this edge server is running, e.g. localhost, Kubernetes
+      // Length must be <= 256 characters.
+      platform: process.env.APOLLO_SERVER_PLATFORM || 'local',
+      runtimeVersion: `node ${process.version}`,
+      executableSchemaId: executableSchemaId,
+      // An identifier used to distinguish the version of the server code such as git or docker sha.
+      // Length must be <= 256 charecters
+      userVersion: process.env.APOLLO_SERVER_USER_VERSION,
+      // "An identifier for the server instance. Length must be <= 256 characters.
+      serverId:
+        process.env.APOLLO_SERVER_ID || process.env.HOSTNAME || os.hostname(),
+      libraryVersion: `apollo-engine-reporting@${
+        require('../package.json').version
+      }`,
+    };
+
+    // Jitter the startup between 0 and 10 seconds
+    const delay = Math.floor(
+      Math.random() *
+        (this.options.experimental_schemaReportingInitialDelayMaxMs || 10_000),
+    );
+
+    const schemaReporter = new SchemaReporter(
+      serverInfo,
+      executableSchema,
+      this.apiKey,
+      this.options.schemaReportingUrl,
+    );
+
+    const fallbackReportingDelayInMs = 20_000;
+
+    this.currentSchemaReporter = schemaReporter;
+    const logger = this.logger;
+
+    setTimeout(function() {
+      reportingLoop(schemaReporter, logger, false, fallbackReportingDelayInMs);
+    }, delay);
+  }
+
   // Stop prevents reports from being sent automatically due to time or buffer
   // size, and stop buffering new traces. You may still manually send a last
   // report by calling sendReport().
@@ -525,6 +671,10 @@ export class EngineReportingAgent<TContext = any> {
       this.reportTimer = undefined;
     }
 
+    if (this.currentSchemaReporter) {
+      this.currentSchemaReporter.stop();
+    }
+
     this.stopped = true;
   }
 
@@ -579,14 +729,14 @@ export class EngineReportingAgent<TContext = any> {
 
   private async sendAllReportsAndReportErrors(): Promise<void> {
     await Promise.all(
-      Object.keys(this.reports).map(schemaHash =>
-        this.sendReportAndReportErrors(schemaHash),
+      Object.keys(this.reports).map(executableSchemaId =>
+        this.sendReportAndReportErrors(executableSchemaId),
       ),
     );
   }
 
-  private sendReportAndReportErrors(schemaHash: string): Promise<void> {
-    return this.sendReport(schemaHash).catch(err => {
+  private sendReportAndReportErrors(executableSchemaId: string): Promise<void> {
+    return this.sendReport(executableSchemaId).catch(err => {
       // This catch block is primarily intended to catch network errors from
       // the retried request itself, which include network errors and non-2xx
       // HTTP errors.
@@ -598,11 +748,11 @@ export class EngineReportingAgent<TContext = any> {
     });
   }
 
-  private resetReport(schemaHash: string) {
-    this.reports[schemaHash] = new Report({
-      header: this.reportHeaders[schemaHash],
+  private resetReport(executableSchemaId: string) {
+    this.reports[executableSchemaId] = new Report({
+      header: this.reportHeaders[executableSchemaId],
     });
-    this.reportSizes[schemaHash] = 0;
+    this.reportSizes[executableSchemaId] = 0;
   }
 }
 
@@ -713,3 +863,15 @@ function makeSendValuesBaseOptionsFromLegacy(
     ? { none: true }
     : { all: true };
 }
+
+export function computeExecutableSchemaId(
+  schema: string | GraphQLSchema,
+): string {
+  // Can't call digest on this object twice. Creating new object each function call
+  const sha256 = createHash('sha256');
+  const schemaDocument =
+    typeof schema === 'string'
+      ? schema
+      : printSchema(schema);
+  return sha256.update(schemaDocument).digest('hex');
+}
diff --git a/packages/apollo-engine-reporting/src/plugin.ts b/packages/apollo-engine-reporting/src/plugin.ts
index 84ead380e19..4b8217813f6 100644
--- a/packages/apollo-engine-reporting/src/plugin.ts
+++ b/packages/apollo-engine-reporting/src/plugin.ts
@@ -5,17 +5,18 @@ import {
   GraphQLRequestContextDidEncounterErrors,
 } from 'apollo-server-types';
 import { Headers } from 'apollo-server-env';
+import { GraphQLSchema, printSchema } from 'graphql';
 import { Trace } from 'apollo-engine-reporting-protobuf';
 
 import {
+  AddTraceArgs,
   EngineReportingOptions,
   GenerateClientInfo,
-  AddTraceArgs,
-  VariableValueOptions,
   SendValuesBaseOptions,
+  VariableValueOptions,
 } from './agent';
 import { EngineReportingTreeBuilder } from './treeBuilder';
-import { ApolloServerPlugin } from "apollo-server-plugin-base";
+import { ApolloServerPlugin } from 'apollo-server-plugin-base';
 
 const clientNameHeaderKey = 'apollographql-client-name';
 const clientReferenceIdHeaderKey = 'apollographql-client-reference-id';
@@ -30,18 +31,39 @@ const clientVersionHeaderKey = 'apollographql-client-version';
 export const plugin = <TContext>(
   options: EngineReportingOptions<TContext> = Object.create(null),
   addTrace: (args: AddTraceArgs) => Promise<void>,
-  // schemaHash: string,
+  {
+    startSchemaReporting,
+    executableSchemaIdGenerator,
+  }: {
+    startSchemaReporting: ({
+      executableSchema,
+      executableSchemaId,
+    }: {
+      executableSchema: string;
+      executableSchemaId: string;
+    }) => void;
+    executableSchemaIdGenerator: (schema: string | GraphQLSchema) => string;
+  },
 ): ApolloServerPlugin<TContext> => {
   const logger: Logger = options.logger || console;
   const generateClientInfo: GenerateClientInfo<TContext> =
     options.generateClientInfo || defaultGenerateClientInfo;
 
-
   return {
+    serverWillStart: function({ schema }) {
+      if (!options.experimental_schemaReporting) return;
+      startSchemaReporting({
+        executableSchema:
+          options.experimental_overrideReportedSchema || printSchema(schema),
+        executableSchemaId: executableSchemaIdGenerator(
+          options.experimental_overrideReportedSchema || schema,
+        ),
+      });
+    },
     requestDidStart({
       logger: requestLogger,
-      schemaHash,
       metrics,
+      schema,
       request: { http, variables },
     }) {
       const treeBuilder: EngineReportingTreeBuilder = new EngineReportingTreeBuilder(
@@ -124,7 +146,9 @@ export const plugin = <TContext>(
           document: requestContext.document,
           source: requestContext.source,
           trace: treeBuilder.trace,
-          schemaHash,
+          executableSchemaId: executableSchemaIdGenerator(
+            options.experimental_overrideReportedSchema || schema,
+          ),
         });
       }
 
@@ -190,7 +214,7 @@ export const plugin = <TContext>(
           didEnd(requestContext);
         },
       };
-    }
+    },
   };
 };
 
diff --git a/packages/apollo-engine-reporting/src/reportingOperationTypes.ts b/packages/apollo-engine-reporting/src/reportingOperationTypes.ts
new file mode 100644
index 00000000000..1bf7b182d51
--- /dev/null
+++ b/packages/apollo-engine-reporting/src/reportingOperationTypes.ts
@@ -0,0 +1,64 @@
+/* tslint:disable */
+/* eslint-disable */
+// @generated
+// This file was automatically generated and should not be edited.
+
+// ====================================================
+// GraphQL mutation operation: AutoregReportServerInfo
+// ====================================================
+
+import { GraphQLFormattedError } from 'graphql';
+
+export interface ReportServerInfo_me_UserMutation {
+  __typename: 'UserMutation';
+}
+
+export interface ReportServerInfo_me_ServiceMutation_reportServerInfo {
+  __typename: 'ReportServerInfoResponse';
+  inSeconds: number;
+  withExecutableSchema: boolean;
+}
+
+export interface ReportServerInfo_me_ServiceMutation {
+  __typename: 'ServiceMutation';
+  /**
+   *  Schema auto-registration. Private alpha.
+   */
+  reportServerInfo: ReportServerInfo_me_ServiceMutation_reportServerInfo | null;
+}
+
+export type ReportServerInfo_me =
+  | ReportServerInfo_me_UserMutation
+  | ReportServerInfo_me_ServiceMutation;
+
+export interface SchemaReportingServerInfo {
+  me: ReportServerInfo_me | null;
+}
+
+export interface SchemaReportingServerInfoResult {
+  data?: SchemaReportingServerInfo;
+  errors?: ReadonlyArray<GraphQLFormattedError>;
+}
+
+export interface ReportServerInfoVariables {
+  info: EdgeServerInfo;
+  executableSchema?: string | null;
+}
+
+/**
+ * Edge server info
+ */
+export interface EdgeServerInfo {
+  bootId: string;
+  executableSchemaId: string;
+  graphVariant: string;
+  libraryVersion?: string | null;
+  platform?: string | null;
+  runtimeVersion?: string | null;
+  serverId?: string | null;
+  userVersion?: string | null;
+}
+
+//==============================================================
+// END Enums and Input Objects
+//==============================================================
diff --git a/packages/apollo-engine-reporting/src/schemaReporter.ts b/packages/apollo-engine-reporting/src/schemaReporter.ts
new file mode 100644
index 00000000000..5dae15d7b61
--- /dev/null
+++ b/packages/apollo-engine-reporting/src/schemaReporter.ts
@@ -0,0 +1,190 @@
+import {
+  ReportServerInfoVariables,
+  EdgeServerInfo,
+  SchemaReportingServerInfoResult,
+} from './reportingOperationTypes';
+import { fetch, Headers, Request } from 'apollo-server-env';
+import { GraphQLRequest, Logger } from 'apollo-server-types';
+
+export const reportServerInfoGql = `
+  mutation ReportServerInfo($info: EdgeServerInfo!, $executableSchema: String) {
+    me {
+      __typename
+      ... on ServiceMutation {
+        reportServerInfo(info: $info, executableSchema: $executableSchema) {
+          inSeconds
+          withExecutableSchema
+        }
+      }
+    }
+  }
+`;
+
+export function reportingLoop(
+  schemaReporter: SchemaReporter,
+  logger: Logger,
+  sendNextWithExecutableSchema: boolean,
+  fallbackReportingDelayInMs: number,
+) {
+  function inner() {
+    // Bail out permanently
+    if (schemaReporter.stopped()) return;
+
+    // Not awaiting this. The callback is handled in the `then` and it calls inner()
+    // to report the server info in however many seconds we were told to wait from
+    // Apollo Graph Manager
+    schemaReporter
+      .reportServerInfo(sendNextWithExecutableSchema)
+      .then(({ inSeconds, withExecutableSchema }) => {
+        sendNextWithExecutableSchema = withExecutableSchema;
+        setTimeout(inner, inSeconds * 1000);
+      })
+      .catch((error: any) => {
+        // In the case of an error we want to continue looping
+        // We can add hardcoded backoff in the future,
+        // or on repeated failures stop responding reporting.
+        logger.error(
+          `Error reporting server info to Apollo Graph Manager during schema reporting: ${error}`,
+        );
+        sendNextWithExecutableSchema = false;
+        setTimeout(inner, fallbackReportingDelayInMs);
+      });
+  }
+
+  inner();
+}
+
+interface ReportServerInfoReturnVal {
+  inSeconds: number;
+  withExecutableSchema: boolean;
+}
+
+// This class is meant to be a thin shim around the gql mutations.
+export class SchemaReporter {
+  // These mirror the gql variables
+  private readonly serverInfo: EdgeServerInfo;
+  private readonly executableSchemaDocument: any;
+  private readonly url: string;
+
+  private isStopped: boolean;
+  private readonly headers: Headers;
+
+  constructor(
+    serverInfo: EdgeServerInfo,
+    schemaSdl: string,
+    apiKey: string,
+    schemaReportingEndpoint: string | undefined,
+  ) {
+    this.headers = new Headers();
+    this.headers.set('Content-Type', 'application/json');
+    this.headers.set('x-api-key', apiKey);
+    this.headers.set('apollographql-client-name', 'apollo-engine-reporting');
+    this.headers.set(
+      'apollographql-client-version',
+      require('../package.json').version,
+    );
+
+    this.url =
+      schemaReportingEndpoint ||
+      'https://engine-graphql.apollographql.com/api/graphql';
+
+    this.serverInfo = serverInfo;
+    this.executableSchemaDocument = schemaSdl;
+    this.isStopped = false;
+  }
+
+  public stopped(): Boolean {
+    return this.isStopped;
+  }
+
+  public stop() {
+    this.isStopped = true;
+  }
+
+  public async reportServerInfo(
+    withExecutableSchema: boolean,
+  ): Promise<ReportServerInfoReturnVal> {
+    const { data, errors } = await this.graphManagerQuery({
+      info: this.serverInfo,
+      executableSchema: withExecutableSchema
+        ? this.executableSchemaDocument
+        : null,
+    });
+
+    if (errors) {
+      throw new Error((errors || []).map((x: any) => x.message).join('\n'));
+    }
+
+    function msgForUnexpectedResponse(data: any): string {
+      return [
+        'Unexpected response shape from Apollo Graph Manager when',
+        'reporting server information for schema reporting. If',
+        'this continues, please reach out to support@apollographql.com.',
+        'Received response:',
+        JSON.stringify(data),
+      ].join(' ');
+    }
+
+    if (!data || !data.me || !data.me.__typename) {
+      throw new Error(msgForUnexpectedResponse(data));
+    }
+
+    if (data.me.__typename === 'UserMutation') {
+      this.isStopped = true;
+      throw new Error(
+        [
+          'This server was configured with an API key for a user.',
+          "Only a service's API key may be used for schema reporting.",
+          'Please visit the settings for this graph at',
+          'https://engine.apollographql.com/ to obtain an API key for a service.',
+        ].join(' '),
+      );
+    } else if (data.me.__typename === 'ServiceMutation') {
+      if (!data.me.reportServerInfo) {
+        throw new Error(msgForUnexpectedResponse(data));
+      }
+      return data.me.reportServerInfo;
+    } else {
+      throw new Error(msgForUnexpectedResponse(data));
+    }
+  }
+
+  private async graphManagerQuery(
+    variables: ReportServerInfoVariables,
+  ): Promise<SchemaReportingServerInfoResult> {
+    const request: GraphQLRequest = {
+      query: reportServerInfoGql,
+      operationName: 'ReportServerInfo',
+      variables: variables,
+    };
+    const httpRequest = new Request(this.url, {
+      method: 'POST',
+      headers: this.headers,
+      body: JSON.stringify(request),
+    });
+
+    const httpResponse = await fetch(httpRequest);
+
+    if (!httpResponse.ok) {
+      throw new Error([
+        `An unexpected HTTP status code (${httpResponse.status}) was`,
+        'encountered during schema reporting.'
+      ].join(' '));
+    }
+
+    try {
+      // JSON parsing failure due to malformed data is the likely failure case
+      // here.  Any non-JSON response (e.g. HTML) is usually the suspect.
+      return await httpResponse.json();
+    } catch (error) {
+      throw new Error(
+        [
+          "Couldn't report server info to Apollo Graph Manager.",
+          'Parsing response as JSON failed.',
+          'If this continues please reach out to support@apollographql.com',
+          error
+        ].join(' '),
+      );
+    }
+  }
+}
diff --git a/packages/apollo-server-core/src/utils/pluginTestHarness.ts b/packages/apollo-server-core/src/utils/pluginTestHarness.ts
index 5c790c70f6c..86dd63460c9 100644
--- a/packages/apollo-server-core/src/utils/pluginTestHarness.ts
+++ b/packages/apollo-server-core/src/utils/pluginTestHarness.ts
@@ -81,7 +81,6 @@ export default async function pluginTestHarness<TContext>({
    */
   context?: TContext;
 }): Promise<GraphQLRequestContextWillSendResponse<TContext>> {
-
   if (!schema) {
     schema = new GraphQLSchema({
       query: new GraphQLObjectType({
@@ -98,6 +97,17 @@ export default async function pluginTestHarness<TContext>({
     });
   }
 
+  const schemaHash = generateSchemaHash(schema);
+  if (typeof pluginInstance.serverWillStart === 'function') {
+    pluginInstance.serverWillStart({
+      logger: logger || console,
+      schema,
+      schemaHash,
+      engine: {},
+    });
+  }
+
+
   const requestContext: GraphQLRequestContext<TContext> = {
     logger: logger || console,
     schema,
diff --git a/packages/apollo-server-integration-testsuite/src/ApolloServer.ts b/packages/apollo-server-integration-testsuite/src/ApolloServer.ts
index 521c52bcabd..85257054458 100644
--- a/packages/apollo-server-integration-testsuite/src/ApolloServer.ts
+++ b/packages/apollo-server-integration-testsuite/src/ApolloServer.ts
@@ -852,7 +852,7 @@ export function testApolloServer<AS extends ApolloServerBase>(
 
           public engineOptions(): Partial<EngineReportingOptions<any>> {
             return {
-              endpointUrl: this.getUrl(),
+              tracesEndpointUrl: this.getUrl(),
             };
           }
 
@@ -2170,7 +2170,7 @@ export function testApolloServer<AS extends ApolloServerBase>(
               resolvers: { Query: { something: () => 'hello' } },
               engine: {
                 apiKey: 'service:my-app:secret',
-                endpointUrl: fakeEngineUrl,
+                tracesEndpointUrl: fakeEngineUrl,
                 reportIntervalMs: 1,
                 maxAttempts: 3,
                 requestAgent,