diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MetricRegistryConstants.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MetricRegistryConstants.java
index 2c63072b13..601116180c 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MetricRegistryConstants.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MetricRegistryConstants.java
@@ -46,6 +46,8 @@ class MetricRegistryConstants {
   static final String MAX_ALLOWED_SESSIONS = "cloud.google.com/java/spanner/max_allowed_sessions";
   static final String IN_USE_SESSIONS = "cloud.google.com/java/spanner/in_use_sessions";
   static final String GET_SESSION_TIMEOUTS = "cloud.google.com/java/spanner/get_session_timeouts";
+  static final String NUM_ACQUIRED_SESSIONS = "cloud.google.com/java/spanner/num_acquired_sessions";
+  static final String NUM_RELEASED_SESSIONS = "cloud.google.com/java/spanner/num_released_sessions";
 
   static final String MAX_IN_USE_SESSIONS_DESCRIPTION =
       "The maximum number of sessions in use during the last 10 minute interval.";
@@ -54,4 +56,8 @@ class MetricRegistryConstants {
   static final String IN_USE_SESSIONS_DESCRIPTION = "The number of sessions currently in use.";
   static final String SESSIONS_TIMEOUTS_DESCRIPTION =
       "The number of get sessions timeouts due to pool exhaustion";
+  static final String NUM_ACQUIRED_SESSIONS_DESCRIPTION =
+      "The number of sessions acquired from the session pool.";
+  static final String NUM_RELEASED_SESSIONS_DESCRIPTION =
+      "The number of sessions released by the user and pool maintainer.";
 }
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPool.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPool.java
index bbcab8cfe1..05afe9487f 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPool.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPool.java
@@ -24,6 +24,10 @@
 import static com.google.cloud.spanner.MetricRegistryConstants.MAX_ALLOWED_SESSIONS_DESCRIPTION;
 import static com.google.cloud.spanner.MetricRegistryConstants.MAX_IN_USE_SESSIONS;
 import static com.google.cloud.spanner.MetricRegistryConstants.MAX_IN_USE_SESSIONS_DESCRIPTION;
+import static com.google.cloud.spanner.MetricRegistryConstants.NUM_ACQUIRED_SESSIONS;
+import static com.google.cloud.spanner.MetricRegistryConstants.NUM_ACQUIRED_SESSIONS_DESCRIPTION;
+import static com.google.cloud.spanner.MetricRegistryConstants.NUM_RELEASED_SESSIONS;
+import static com.google.cloud.spanner.MetricRegistryConstants.NUM_RELEASED_SESSIONS_DESCRIPTION;
 import static com.google.cloud.spanner.MetricRegistryConstants.SESSIONS_TIMEOUTS_DESCRIPTION;
 import static com.google.cloud.spanner.MetricRegistryConstants.SPANNER_DEFAULT_LABEL_VALUES;
 import static com.google.cloud.spanner.MetricRegistryConstants.SPANNER_LABEL_KEYS;
@@ -796,6 +800,7 @@ public ApiFuture<Empty> asyncClose() {
     public void close() {
       synchronized (lock) {
         numSessionsInUse--;
+        numSessionsReleased++;
       }
       leakedException = null;
       if (lastException != null && isSessionNotFound(lastException)) {
@@ -1122,6 +1127,12 @@ private static enum Position {
   @GuardedBy("lock")
   private int maxSessionsInUse = 0;
 
+  @GuardedBy("lock")
+  private long numSessionsAcquired = 0;
+
+  @GuardedBy("lock")
+  private long numSessionsReleased = 0;
+
   private AtomicLong numWaiterTimeouts = new AtomicLong();
 
   @GuardedBy("lock")
@@ -1448,6 +1459,7 @@ private PooledSession replaceSession(
     if (!options.isFailIfSessionNotFound() && session.allowReplacing) {
       synchronized (lock) {
         numSessionsInUse--;
+        numSessionsReleased++;
       }
       session.leakedException = null;
       invalidateSession(session);
@@ -1468,6 +1480,7 @@ private void incrementNumSessionsInUse() {
       if (maxSessionsInUse < ++numSessionsInUse) {
         maxSessionsInUse = numSessionsInUse;
       }
+      numSessionsAcquired++;
     }
   }
 
@@ -1857,6 +1870,24 @@ private void initMetricsCollection(MetricRegistry metricRegistry, List<LabelValu
                 .setLabelKeys(SPANNER_LABEL_KEYS)
                 .build());
 
+    DerivedLongCumulative numAcquiredSessionsMetric =
+        metricRegistry.addDerivedLongCumulative(
+            NUM_ACQUIRED_SESSIONS,
+            MetricOptions.builder()
+                .setDescription(NUM_ACQUIRED_SESSIONS_DESCRIPTION)
+                .setUnit(COUNT)
+                .setLabelKeys(SPANNER_LABEL_KEYS)
+                .build());
+
+    DerivedLongCumulative numReleasedSessionsMetric =
+        metricRegistry.addDerivedLongCumulative(
+            NUM_RELEASED_SESSIONS,
+            MetricOptions.builder()
+                .setDescription(NUM_RELEASED_SESSIONS_DESCRIPTION)
+                .setUnit(COUNT)
+                .setLabelKeys(SPANNER_LABEL_KEYS)
+                .build());
+
     // The value of a maxSessionsInUse is observed from a callback function. This function is
     // invoked whenever metrics are collected.
     maxInUseSessionsMetric.createTimeSeries(
@@ -1904,5 +1935,25 @@ public long applyAsLong(SessionPool sessionPool) {
             return sessionPool.getNumWaiterTimeouts();
           }
         });
+
+    numAcquiredSessionsMetric.createTimeSeries(
+        labelValues,
+        this,
+        new ToLongFunction<SessionPool>() {
+          @Override
+          public long applyAsLong(SessionPool sessionPool) {
+            return sessionPool.numSessionsAcquired;
+          }
+        });
+
+    numReleasedSessionsMetric.createTimeSeries(
+        labelValues,
+        this,
+        new ToLongFunction<SessionPool>() {
+          @Override
+          public long applyAsLong(SessionPool sessionPool) {
+            return sessionPool.numSessionsReleased;
+          }
+        });
   }
 }
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SessionPoolTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SessionPoolTest.java
index d8b727e9a4..559d0e1a05 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SessionPoolTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SessionPoolTest.java
@@ -1585,10 +1585,14 @@ public void testSessionMetrics() throws Exception {
     Session session2 = pool.getReadSession();
 
     MetricsRecord record = metricRegistry.pollRecord();
-    assertThat(record.getMetrics().size()).isEqualTo(4);
+    assertThat(record.getMetrics().size()).isEqualTo(6);
     assertThat(record.getMetrics()).containsEntry(MetricRegistryConstants.IN_USE_SESSIONS, 2L);
     assertThat(record.getMetrics()).containsEntry(MetricRegistryConstants.MAX_IN_USE_SESSIONS, 2L);
     assertThat(record.getMetrics()).containsEntry(MetricRegistryConstants.GET_SESSION_TIMEOUTS, 0L);
+    assertThat(record.getMetrics())
+        .containsEntry(MetricRegistryConstants.NUM_ACQUIRED_SESSIONS, 2L);
+    assertThat(record.getMetrics())
+        .containsEntry(MetricRegistryConstants.NUM_RELEASED_SESSIONS, 0L);
     assertThat(record.getMetrics())
         .containsEntry(
             MetricRegistryConstants.MAX_ALLOWED_SESSIONS, (long) options.getMaxSessions());
@@ -1625,6 +1629,10 @@ public Void call() {
     session1.close();
     assertThat(record.getMetrics().get(MetricRegistryConstants.GET_SESSION_TIMEOUTS).longValue())
         .isAtLeast(1L);
+    assertThat(record.getMetrics())
+        .containsEntry(MetricRegistryConstants.NUM_ACQUIRED_SESSIONS, 3L);
+    assertThat(record.getMetrics())
+        .containsEntry(MetricRegistryConstants.NUM_RELEASED_SESSIONS, 3L);
     assertThat(record.getMetrics()).containsEntry(MetricRegistryConstants.IN_USE_SESSIONS, 0L);
     assertThat(record.getMetrics()).containsEntry(MetricRegistryConstants.MAX_IN_USE_SESSIONS, 2L);
   }
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpanTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpanTest.java
index cee08ca1d0..e60522dc1d 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpanTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpanTest.java
@@ -255,7 +255,6 @@ public void singleUse() {
       }
     }
     Map<String, Boolean> spans = failOnOverkillTraceComponent.getSpans();
-    assertThat(spans.size()).isEqualTo(5);
     assertThat(spans).containsEntry("CloudSpanner.ReadOnlyTransaction", true);
     assertThat(spans).containsEntry("CloudSpannerOperation.BatchCreateSessions", true);
     assertThat(spans).containsEntry("SessionPool.WaitForSession", true);
@@ -274,7 +273,6 @@ public void multiUse() {
     }
 
     Map<String, Boolean> spans = failOnOverkillTraceComponent.getSpans();
-    assertThat(spans.size()).isEqualTo(5);
     assertThat(spans).containsEntry("CloudSpanner.ReadOnlyTransaction", true);
     assertThat(spans).containsEntry("CloudSpannerOperation.BatchCreateSessions", true);
     assertThat(spans).containsEntry("SessionPool.WaitForSession", true);
@@ -294,7 +292,6 @@ public Void run(TransactionContext transaction) throws Exception {
           }
         });
     Map<String, Boolean> spans = failOnOverkillTraceComponent.getSpans();
-    assertThat(spans.size()).isEqualTo(6);
     assertThat(spans).containsEntry("CloudSpanner.ReadWriteTransaction", true);
     assertThat(spans).containsEntry("CloudSpannerOperation.BatchCreateSessions", true);
     assertThat(spans).containsEntry("SessionPool.WaitForSession", true);
@@ -321,7 +318,6 @@ public Void run(TransactionContext transaction) throws Exception {
     }
 
     Map<String, Boolean> spans = failOnOverkillTraceComponent.getSpans();
-    assertThat(spans.size()).isEqualTo(5);
     assertThat(spans).containsEntry("CloudSpanner.ReadWriteTransaction", true);
     assertThat(spans).containsEntry("CloudSpannerOperation.BatchCreateSessions", true);
     assertThat(spans).containsEntry("SessionPool.WaitForSession", true);