diff --git a/api/src/main/java/com/google/common/flogger/LogContext.java b/api/src/main/java/com/google/common/flogger/LogContext.java
index dbae7192..08d8c945 100644
--- a/api/src/main/java/com/google/common/flogger/LogContext.java
+++ b/api/src/main/java/com/google/common/flogger/LogContext.java
@@ -100,7 +100,7 @@ public void emitRepeated(Iterator<Object> keys, KeyValueHandler out) {
             if (!keys.hasNext()) {
               out.handle(getLabel(), first);
             } else {
-              // Or we could emit the .
+              // In the very unlikely case there's more than one aggregation key, emit a list.
               StringBuilder buf = new StringBuilder();
               buf.append('[').append(first);
               do {
@@ -583,6 +583,10 @@ private boolean shouldLog() {
     LogSiteKey logSiteKey = null;
     if (logSite != LogSite.INVALID) {
       logSiteKey = logSite;
+      // Log site keys are only modified when we have metadata in the log statement.
+      if (metadata != null && metadata.size() > 0) {
+        logSiteKey = specializeLogSiteKeyFromMetadata(logSiteKey, metadata);
+      }
     }
     if (!postProcess(logSiteKey)) {
       return false;
@@ -603,8 +607,11 @@ private boolean shouldLog() {
   // that a log statement executed both in and outside of the context would currently see
   // different keys, when they should be the same. To fix this, specialization must be changed
   // to ignore repeated scopes. For now we only see log site metadata so this is not an issue.
+  //
   // TODO: Ignore repeated scopes (e.g. use a Bloom Filter mask on each scope).
   // TODO: Make a proper iterator on Metadata or use MetadataProcessor.
+  //
+  // Visible for testing
   static LogSiteKey specializeLogSiteKeyFromMetadata(LogSiteKey logSiteKey, Metadata metadata) {
     checkNotNull(logSiteKey, "logSiteKey"); // For package null checker only.
     for (int n = 0, size = metadata.size(); n < size; n++) {
@@ -701,16 +708,24 @@ public final <T> API with(MetadataKey<Boolean> key) {
     return with(key, Boolean.TRUE);
   }
 
+  @Override
+  public final API per(Enum<?> key) {
+    return with(Key.LOG_SITE_GROUPING_KEY, key);
+  }
+
+  @Override
+  public API per(LoggingScopeProvider scopeProvider) {
+    return with(Key.LOG_SITE_GROUPING_KEY, scopeProvider.getCurrentScope());
+  }
+
   @Override
   public final API withCause(Throwable cause) {
-    if (cause != null) {
-      addMetadata(Key.LOG_CAUSE, cause);
-    }
-    return api();
+    return with(Key.LOG_CAUSE, cause);
   }
 
   @Override
   public API withStackTrace(StackSize size) {
+    // Unlike other metadata methods, the "no-op" value is not null.
     if (checkNotNull(size, "stack size") != StackSize.NONE) {
       addMetadata(Key.CONTEXT_STACK_SIZE, size);
     }
diff --git a/api/src/main/java/com/google/common/flogger/LoggingApi.java b/api/src/main/java/com/google/common/flogger/LoggingApi.java
index 99e60117..a9480819 100644
--- a/api/src/main/java/com/google/common/flogger/LoggingApi.java
+++ b/api/src/main/java/com/google/common/flogger/LoggingApi.java
@@ -124,6 +124,78 @@ public interface LoggingApi<API extends LoggingApi<API>> {
    */
   API atMostEvery(int n, TimeUnit unit);
 
+  /**
+   * Aggregates stateful logging with respect to the given enum value.
+   *
+   * <p>Normally log statements with conditional behaviour (e.g. rate limiting) use the same state
+   * for each invocation (e.g. counters or timestamps). This method allows an additional qualifier
+   * to be given which allows for different conditional state for each unique qualifier.
+   *
+   * <p>This only makes a difference for log statements which use persistent state to control
+   * conditional behaviour (e.g. {@code atMostEvery()} or {@code every()}).
+   *
+   * <p>This method is most useful in helping to avoid cases where a rare event might never be
+   * logged due to rate limiting. For example, the following code will cause log statements with
+   * different {@code taskType}s to be rate-limited independently of each other.
+   *
+   * <pre>{@code
+   * // We want to rate limit logging separately for all task types.
+   * logger.at(INFO).per(taskType).atMostEvery(30, SECONDS).log("Start task: %s", taskSpec);
+   * }</pre>
+   *
+   * <p>The {@code key} passed to this method should always be a variable (passing a constant value
+   * has no effect). If {@code null} is passed, this call has no effect (e.g. rate limiting will
+   * apply normally, without respect to any specific scope).
+   *
+   * <p>If multiple aggregation keys are added to a single log statement, then they all take effect
+   * and logging is aggregated by the unique combination of keys passed to all "per" methods.
+   */
+  API per(@NullableDecl Enum<?> key);
+
+  /**
+   * Aggregates stateful logging with respect to a scoped context determined by the given scope
+   * provider.
+   *
+   * <p>When {@link com.google.common.flogger.context.ScopedLoggingContext ScopedLoggingContext} is
+   * used to create a context, it can be bound to a {@link
+   * com.google.common.flogger.context.ScopeType ScopeType}. For example:
+   *
+   * <pre>{@code
+   * ScopedLoggingContexts.newContext(REQUEST).run(() -> scopedMethod(x, y, z));
+   * }</pre>
+   *
+   * where {@link com.google.common.flogger.context.ScopeType#REQUEST REQUEST} defines the scope
+   * type for the context in which {@code scopedMethod()} is called. Within this context, the scope
+   * associated with the {@code REQUEST} type can then be used to aggregate logging behavior:
+   *
+   * <pre>{@code
+   * logger.atInfo().atMostEvery(5, SECONDS).per(REQUEST).log("Some message...");
+   * }</pre>
+   *
+   * <p>New scope types can be created for specific subtasks using {@link
+   * com.google.common.flogger.context.ScopeType#create ScopeType.create("<name>")} but it is
+   * recommended to use shared constants (such as {@code ScopeType.REQUEST}) wherever feasible to
+   * avoid confusion.
+   *
+   * <p>Note that in order for the request scope to be applied to a log statement, the {@code
+   * per(REQUEST)} method must still be called; just being inside the request scope isn't enough.
+   *
+   * <p>Unlike other {@code per()} methods, this method is expected to be given a constant value.
+   * This is because the given value <em>provides</em> the current scope, rather than <em>being</em>
+   * the current scope.
+   *
+   * <p>If a log statement using this method is invoked outside a context of the given type, this
+   * call has no effect (e.g. rate limiting will apply normally, without respect to any specific
+   * scope).
+   *
+   * <p>If multiple aggregation keys are added to a single log statement, then they all take effect
+   * and logging is aggregated by the unique combination of keys passed to all "per" methods.
+   *
+   * @param scopeProvider a constant used to defined the type of the scope in which logging is
+   *     aggregated.
+   */
+  API per(LoggingScopeProvider scopeProvider);
+
   /**
    * Creates a synthetic exception and attaches it as the "cause" of the log statement as a way to
    * provide additional context for the logging call itself. The exception created by this method is
@@ -133,11 +205,6 @@ public interface LoggingApi<API extends LoggingApi<API>> {
    * after) {@code withStackTrace()}, the given exception becomes the cause of the synthetic
    * exception.
    *
-   * <p>Note that this method is experimental and may change in the future (using a "cause" to
-   * provide additional debugging for normal log statements seems hacky and once ECatcher and other
-   * tools can process Flogger's data in a more structured way, there should be no need to tunnel
-   * the metadata via the cause).
-   *
    * @param size the maximum size of the stack trace to be generated.
    */
   API withStackTrace(StackSize size);
@@ -679,6 +746,16 @@ public final <T> API with(MetadataKey<Boolean> key) {
       return noOp();
     }
 
+    @Override
+    public API per(@NullableDecl Enum<?> key) {
+      return noOp();
+    }
+
+    @Override
+    public API per(LoggingScopeProvider scopeProvider) {
+      return noOp();
+    }
+
     @Override
     public final API withCause(@NullableDecl Throwable cause) {
       return noOp();
diff --git a/api/src/main/java/com/google/common/flogger/backend/system/LoggingContext.java b/api/src/main/java/com/google/common/flogger/backend/system/LoggingContext.java
index 8ac0f1d5..2e5a9b8e 100644
--- a/api/src/main/java/com/google/common/flogger/backend/system/LoggingContext.java
+++ b/api/src/main/java/com/google/common/flogger/backend/system/LoggingContext.java
@@ -18,6 +18,7 @@
 
 import com.google.common.flogger.context.ContextDataProvider;
 import com.google.common.flogger.context.LogLevelMap;
+import com.google.common.flogger.context.ScopeType;
 import com.google.common.flogger.context.ScopedLoggingContext;
 import com.google.common.flogger.context.ScopedLoggingContext.LoggingContextCloseable;
 import com.google.common.flogger.context.Tags;
@@ -47,6 +48,12 @@ public LoggingContextCloseable install() {
       };
     }
 
+    @Override
+    public Builder newContext(ScopeType scopeType) {
+      // Scopes unsupported in the old LoggingContext based implementations.
+      return newContext();
+    }
+
     @Override
     public void close() {}
 
diff --git a/api/src/main/java/com/google/common/flogger/context/NoOpContextDataProvider.java b/api/src/main/java/com/google/common/flogger/context/NoOpContextDataProvider.java
index 9245e7f9..e3c3bd78 100644
--- a/api/src/main/java/com/google/common/flogger/context/NoOpContextDataProvider.java
+++ b/api/src/main/java/com/google/common/flogger/context/NoOpContextDataProvider.java
@@ -70,6 +70,12 @@ public LoggingContextCloseable install() {
       };
     }
 
+    @Override
+    public ScopedLoggingContext.Builder newContext(ScopeType scopeType) {
+      // Ignore scope bindings when there's no way to propagate them.
+      return newContext();
+    }
+
     @Override
     public boolean addTags(Tags tags) {
       logWarningOnceOnly();
diff --git a/api/src/main/java/com/google/common/flogger/context/ScopedLoggingContext.java b/api/src/main/java/com/google/common/flogger/context/ScopedLoggingContext.java
index 99573249..fee1609c 100644
--- a/api/src/main/java/com/google/common/flogger/context/ScopedLoggingContext.java
+++ b/api/src/main/java/com/google/common/flogger/context/ScopedLoggingContext.java
@@ -350,6 +350,22 @@ protected ScopedLoggingContext() {}
   @CheckReturnValue
   public abstract Builder newContext();
 
+  /**
+   * Creates a new context builder to which additional logging metadata can be attached before being
+   * installed or used to wrap some existing code.
+   *
+   * <p>This method is the same as {@link #newContext()} except it additionally binds a new {@link
+   * ScopeType} instance to the newly created context. This allows log statements to control
+   * stateful logging operations (e.g. rate limiting) using the {@link
+   * com.google.common.flogger.LoggingApi#per(ScopeType) per(ScopeType)} method.
+   *
+   * <p>Note for users: if you don't need an instance of {@code ScopedLoggingContext} for some
+   * reason such as testability (injecting it, for example), consider using the static methods in
+   * {@link ScopedLoggingContexts} instead to avoid the need to call {@link #getInstance}.
+   */
+  @CheckReturnValue
+  public abstract Builder newContext(ScopeType scopeType);
+
   /**
    * Deprecated equivalent to {@link #newContext()}.
    *
diff --git a/api/src/test/java/com/google/common/flogger/LogContextTest.java b/api/src/test/java/com/google/common/flogger/LogContextTest.java
index f58d561e..7ad06dad 100644
--- a/api/src/test/java/com/google/common/flogger/LogContextTest.java
+++ b/api/src/test/java/com/google/common/flogger/LogContextTest.java
@@ -17,6 +17,8 @@
 package com.google.common.flogger;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.flogger.LogContextTest.LogType.BAR;
+import static com.google.common.flogger.LogContextTest.LogType.FOO;
 import static com.google.common.flogger.LogSites.logSite;
 import static com.google.common.truth.Correspondence.transforming;
 import static com.google.common.truth.Truth.assertThat;
@@ -55,6 +57,9 @@ public class LogContextTest {
   private static final MetadataKey<String> REPEATED_KEY = MetadataKey.repeated("str", String.class);
   private static final MetadataKey<Boolean> FLAG_KEY = MetadataKey.repeated("flag", Boolean.class);
 
+  private static final LogSiteStats.RateLimitPeriod ONCE_PER_SECOND =
+      LogSiteStats.newRateLimitPeriod(1, SECONDS);
+
   @Test
   public void testIsEnabled() {
     FakeLoggerBackend backend = new FakeLoggerBackend();
@@ -193,6 +198,73 @@ public void testAtMostEvery() {
     backend.assertLogged(2).hasArguments(8);
   }
 
+  // Non-private to allow static import to keep test code concise.
+  enum LogType {
+    FOO,
+    BAR
+  };
+
+  @Test
+  public void testPer_enum() {
+    FakeLoggerBackend backend = new FakeLoggerBackend();
+    TestLogger logger = TestLogger.create(backend);
+
+    // Logs for both types should appear (even though the 2nd log is within the rate limit period).
+    // NOTE: It is important this is tested on a single log statement.
+    long nowNanos = MILLISECONDS.toNanos(System.currentTimeMillis());
+    for (LogType type : Arrays.asList(FOO, FOO, FOO, BAR, FOO, BAR, FOO)) {
+      logger.at(INFO, nowNanos).atMostEvery(1, SECONDS).per(type).log("Type: %s", type);
+      nowNanos += MILLISECONDS.toNanos(100);
+    }
+
+    assertThat(backend.getLoggedCount()).isEqualTo(2);
+
+    // Rate limit period and the aggregation key from "per"
+    backend.assertLogged(0).hasArguments(FOO);
+    backend.assertLogged(0).metadata().hasSize(2);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_SITE_GROUPING_KEY, FOO);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_AT_MOST_EVERY, ONCE_PER_SECOND);
+
+    backend.assertLogged(1).hasArguments(BAR);
+    backend.assertLogged(1).metadata().hasSize(2);
+    backend.assertLogged(1).metadata().containsUniqueEntry(Key.LOG_SITE_GROUPING_KEY, BAR);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_AT_MOST_EVERY, ONCE_PER_SECOND);
+  }
+
+  @Test
+  public void testPer_scopeProvider() {
+    FakeLoggerBackend backend = new FakeLoggerBackend();
+    TestLogger logger = TestLogger.create(backend);
+
+    // We can't test a specific implementation of ScopedLoggingContext here (there might not be one
+    // available), so we fake it. The ScopedLoggingContext behaviour is well tested elsewhere. Only
+    // tests should ever create "immediate providers" like this as it doesn't make sense otherwise.
+    LoggingScope fooScope = LoggingScope.create("foo");
+    LoggingScope barScope = LoggingScope.create("bar");
+    LoggingScopeProvider foo = () -> fooScope;
+    LoggingScopeProvider bar = () -> barScope;
+
+    // Logs for both scopes should appear (even though the 2nd log is within the rate limit period).
+    // NOTE: It is important this is tested on a single log statement.
+    long nowNanos = MILLISECONDS.toNanos(System.currentTimeMillis());
+    for (LoggingScopeProvider sp : Arrays.asList(foo, foo, foo, bar, foo, bar, foo)) {
+      logger.at(INFO, nowNanos).atMostEvery(1, SECONDS).per(sp).log("message");
+      nowNanos += MILLISECONDS.toNanos(100);
+    }
+
+    assertThat(backend.getLoggedCount()).isEqualTo(2);
+
+    // Rate limit period and the aggregation key from "per"
+    backend.assertLogged(0).metadata().hasSize(2);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_SITE_GROUPING_KEY, fooScope);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_AT_MOST_EVERY, ONCE_PER_SECOND);
+
+    backend.assertLogged(1).metadata().hasSize(2);
+    backend.assertLogged(1).metadata().containsUniqueEntry(Key.LOG_SITE_GROUPING_KEY, barScope);
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_AT_MOST_EVERY, ONCE_PER_SECOND);
+  }
+
+
   @Test
   public void testWasForced_level() {
     FakeLoggerBackend backend = new FakeLoggerBackend();
@@ -265,10 +337,7 @@ public void testWasForced_atMostEvery() {
     assertThat(backend.getLoggedCount()).isEqualTo(2);
     backend.assertLogged(0).hasMessage("LOGGED 1");
     backend.assertLogged(0).metadata().hasSize(1);
-    backend
-        .assertLogged(0)
-        .metadata()
-        .containsUniqueEntry(Key.LOG_AT_MOST_EVERY, LogSiteStats.newRateLimitPeriod(1, SECONDS));
+    backend.assertLogged(0).metadata().containsUniqueEntry(Key.LOG_AT_MOST_EVERY, ONCE_PER_SECOND);
 
     backend.assertLogged(1).hasMessage("LOGGED 2");
     backend.assertLogged(1).metadata().hasSize(1);
diff --git a/api/src/test/java/com/google/common/flogger/context/ScopedLoggingContextTest.java b/api/src/test/java/com/google/common/flogger/context/ScopedLoggingContextTest.java
index 0d2309eb..dc184212 100644
--- a/api/src/test/java/com/google/common/flogger/context/ScopedLoggingContextTest.java
+++ b/api/src/test/java/com/google/common/flogger/context/ScopedLoggingContextTest.java
@@ -44,6 +44,11 @@ public LoggingContextCloseable install() {
           };
         }
 
+        @Override
+        public Builder newContext(ScopeType scopeType) {
+          return newContext();
+        }
+
         @Override
         public boolean applyLogLevelMap(LogLevelMap m) {
           return false;
diff --git a/api/src/test/java/com/google/common/flogger/testing/AbstractScopedLoggingContextTest.java b/api/src/test/java/com/google/common/flogger/testing/AbstractScopedLoggingContextTest.java
index 52c8543a..122c98a2 100644
--- a/api/src/test/java/com/google/common/flogger/testing/AbstractScopedLoggingContextTest.java
+++ b/api/src/test/java/com/google/common/flogger/testing/AbstractScopedLoggingContextTest.java
@@ -22,6 +22,7 @@
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.flogger.MetadataKey;
+import com.google.common.flogger.LoggingScope;
 import com.google.common.flogger.backend.Metadata;
 import com.google.common.flogger.context.ContextDataProvider;
 import com.google.common.flogger.context.LogLevelMap;
@@ -239,4 +240,58 @@ public void testNewContext_withMergedLevelMap() {
     assertLogging("foo.bar.Baz", Level.FINE).isFalse();
     checkDone();
   }
+
+  @Test
+  public void testNewContext_withBoundScopeTypes() {
+    assertThat(dataProvider.getScope(SUB_TASK)).isNull();
+    assertThat(dataProvider.getScope(BATCH_JOB)).isNull();
+    context
+        .newContext(SUB_TASK)
+        .run(
+            () -> {
+              LoggingScope taskScope = dataProvider.getScope(SUB_TASK);
+              assertThat(taskScope).isNotNull();
+              assertThat(taskScope.toString()).isEqualTo("sub task");
+              assertThat(dataProvider.getScope(BATCH_JOB)).isNull();
+              context
+                  .newContext(BATCH_JOB)
+                  .run(
+                      () -> {
+                        assertThat(dataProvider.getScope(SUB_TASK)).isSameInstanceAs(taskScope);
+                        assertThat(dataProvider.getScope(BATCH_JOB)).isNotNull();
+                        markTestAsDone();
+                      });
+              // Everything is restored after a scope.
+              assertThat(dataProvider.getScope(SUB_TASK)).isSameInstanceAs(taskScope);
+              assertThat(dataProvider.getScope(BATCH_JOB)).isNull();
+            });
+    // Everything is restored after a scope.
+    assertThat(dataProvider.getScope(SUB_TASK)).isNull();
+    assertThat(dataProvider.getScope(BATCH_JOB)).isNull();
+    checkDone();
+  }
+
+  @Test
+  public void testNewContext_repeatedScopesAreIdempotent() {
+    assertThat(dataProvider.getScope(SUB_TASK)).isNull();
+    context
+        .newContext(SUB_TASK)
+        .run(
+            () -> {
+              LoggingScope taskScope = dataProvider.getScope(SUB_TASK);
+              assertThat(taskScope).isNotNull();
+              assertThat(taskScope.toString()).isEqualTo("sub task");
+              context
+                  .newContext(SUB_TASK)
+                  .run(
+                      () -> {
+                        // We don't make a new scope instance if the same type is bound twice!
+                        assertThat(dataProvider.getScope(SUB_TASK)).isSameInstanceAs(taskScope);
+                        markTestAsDone();
+                      });
+              assertThat(dataProvider.getScope(SUB_TASK)).isNotNull();
+            });
+    assertThat(dataProvider.getScope(SUB_TASK)).isNull();
+    checkDone();
+  }
 }
diff --git a/google/src/main/java/com/google/common/flogger/GoogleLoggingApi.java b/google/src/main/java/com/google/common/flogger/GoogleLoggingApi.java
index 4e7baad4..a957f509 100644
--- a/google/src/main/java/com/google/common/flogger/GoogleLoggingApi.java
+++ b/google/src/main/java/com/google/common/flogger/GoogleLoggingApi.java
@@ -36,6 +36,5 @@ public interface GoogleLoggingApi<API extends GoogleLoggingApi<API>> extends Log
    * An implementation of {@link GoogleLoggingApi} which does nothing and discards all parameters.
    */
   public static class NoOp<API extends GoogleLoggingApi<API>> extends LoggingApi.NoOp<API>
-      implements GoogleLoggingApi<API> {
-  }
+      implements GoogleLoggingApi<API> {}
 }
diff --git a/grpc/src/main/java/com/google/common/flogger/grpc/GrpcContextDataProvider.java b/grpc/src/main/java/com/google/common/flogger/grpc/GrpcContextDataProvider.java
index 21ac8c62..d2277143 100644
--- a/grpc/src/main/java/com/google/common/flogger/grpc/GrpcContextDataProvider.java
+++ b/grpc/src/main/java/com/google/common/flogger/grpc/GrpcContextDataProvider.java
@@ -16,8 +16,10 @@
 
 package com.google.common.flogger.grpc;
 
+import com.google.common.flogger.LoggingScope;
 import com.google.common.flogger.context.ContextDataProvider;
 import com.google.common.flogger.context.ContextMetadata;
+import com.google.common.flogger.context.ScopeType;
 import com.google.common.flogger.context.ScopedLoggingContext;
 import com.google.common.flogger.context.Tags;
 import io.grpc.Context;
@@ -82,6 +84,12 @@ public ContextMetadata getMetadata() {
     return GrpcContextData.getMetadataFor(currentContext());
   }
 
+  @NullableDecl
+  @Override
+  public LoggingScope getScope(ScopeType type) {
+    return GrpcContextData.lookupScopeFor(currentContext(), type);
+  }
+
   @Override
   public boolean shouldForceLogging(String loggerName, Level level, boolean isEnabledByLevel) {
     // Shortcutting boolean saves doing any work in the commonest case (this code is called for
diff --git a/grpc/src/main/java/com/google/common/flogger/grpc/GrpcScopedLoggingContext.java b/grpc/src/main/java/com/google/common/flogger/grpc/GrpcScopedLoggingContext.java
index dc65f021..1ea963d6 100644
--- a/grpc/src/main/java/com/google/common/flogger/grpc/GrpcScopedLoggingContext.java
+++ b/grpc/src/main/java/com/google/common/flogger/grpc/GrpcScopedLoggingContext.java
@@ -49,6 +49,12 @@ public ScopedLoggingContext.Builder newContext() {
     return newBuilder(null);
   }
 
+  @Override
+  @CheckReturnValue
+  public ScopedLoggingContext.Builder newContext(ScopeType scopeType) {
+    return newBuilder(scopeType);
+  }
+
   private ScopedLoggingContext.Builder newBuilder(@NullableDecl ScopeType scopeType) {
     return new ScopedLoggingContext.Builder() {
       @Override