diff --git a/google/cloud/logging_v2/__init__.py b/google/cloud/logging_v2/__init__.py
index 98954d550..d55e474d9 100644
--- a/google/cloud/logging_v2/__init__.py
+++ b/google/cloud/logging_v2/__init__.py
@@ -41,6 +41,8 @@
 """Query string to order by ascending timestamps."""
 DESCENDING = "timestamp desc"
 """Query string to order by decending timestamps."""
+_instrumentation_emitted = False
+"""Flag for whether instrumentation info has been emitted"""
 
 
 __all__ = (
diff --git a/google/cloud/logging_v2/_instrumentation.py b/google/cloud/logging_v2/_instrumentation.py
new file mode 100644
index 000000000..0d9de76d3
--- /dev/null
+++ b/google/cloud/logging_v2/_instrumentation.py
@@ -0,0 +1,90 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Add diagnostic instrumentation source information to logs"""
+from google.cloud.logging_v2.entries import StructEntry
+from google.cloud.logging_v2 import __version__
+
+_DIAGNOSTIC_INFO_KEY = "logging.googleapis.com/diagnostic"
+_INSTRUMENTATION_SOURCE_KEY = "instrumentation_source"
+_PYTHON_LIBRARY_NAME = "python"
+
+_LIBRARY_VERSION = __version__
+
+_MAX_NAME_LENGTH = 14
+_MAX_VERSION_LENGTH = 14
+_MAX_INSTRUMENTATION_ENTRIES = 3
+
+
+def _add_instrumentation(entries, **kw):
+    """Add instrumentation information to a list of entries
+
+        A new diagnostic entry is prepended to the list of
+        entries.
+
+    Args:
+       entries (Sequence[Mapping[str, ...]]): sequence of mappings representing
+            the log entry resources to log.
+
+    Returns:
+        Sequence[Mapping[str, ...]]: entries with instrumentation info added to
+        the beginning of list.
+    """
+
+    diagnostic_entry = _create_diagnostic_entry(**kw)
+    entries.insert(0, diagnostic_entry.to_api_repr())
+    return entries
+
+
+def _create_diagnostic_entry(name=_PYTHON_LIBRARY_NAME, version=_LIBRARY_VERSION, **kw):
+    """Create a diagnostic log entry describing this library
+
+        The diagnostic log consists of a list of library name and version objects
+        that have handled a given log entry.  If this library is the originator
+        of the log entry, it will look like:
+        {logging.googleapis.com/diagnostic: {instrumentation_source: [{name: "python", version: "3.0.0"}]}}
+
+    Args:
+        name(str): The name of this library (e.g. 'python')
+        version(str) The version of this library (e.g. '3.0.0')
+
+    Returns:
+        google.cloud.logging_v2.LogEntry: Log entry with library information
+    """
+    payload = {
+        _DIAGNOSTIC_INFO_KEY: {
+            _INSTRUMENTATION_SOURCE_KEY: [_get_instrumentation_source(name, version)]
+        }
+    }
+    kw["severity"] = "INFO"
+    entry = StructEntry(payload=payload, **kw)
+    return entry
+
+
+def _get_instrumentation_source(name=_PYTHON_LIBRARY_NAME, version=_LIBRARY_VERSION):
+    """Gets a JSON representation of the instrumentation_source
+
+    Args:
+        name(str): The name of this library (e.g. 'python')
+        version(str) The version of this library (e.g. '3.0.0')
+    Returns:
+       obj: JSON object with library information
+    """
+    source = {"name": name, "version": version}
+    # truncate strings to no more than _MAX_NAME_LENGTH characters
+    for key, val in source.items():
+        source[key] = (
+            val if len(val) <= _MAX_NAME_LENGTH else f"{val[:_MAX_NAME_LENGTH]}*"
+        )
+    return source
diff --git a/google/cloud/logging_v2/handlers/structured_log.py b/google/cloud/logging_v2/handlers/structured_log.py
index 2d7c5e078..4a9a139e5 100644
--- a/google/cloud/logging_v2/handlers/structured_log.py
+++ b/google/cloud/logging_v2/handlers/structured_log.py
@@ -16,10 +16,13 @@
 """
 import collections
 import json
+import logging
 import logging.handlers
 
 from google.cloud.logging_v2.handlers.handlers import CloudLoggingFilter
 from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message
+import google.cloud.logging_v2
+from google.cloud.logging_v2._instrumentation import _create_diagnostic_entry
 
 GCP_FORMAT = (
     "{%(_payload_str)s"
@@ -84,3 +87,13 @@ def format(self, record):
         # convert to GCP structred logging format
         gcp_payload = self._gcp_formatter.format(record)
         return gcp_payload
+
+    def emit(self, record):
+        if google.cloud.logging_v2._instrumentation_emitted is False:
+            self.emit_instrumentation_info()
+        super().emit(record)
+
+    def emit_instrumentation_info(self):
+        google.cloud.logging_v2._instrumentation_emitted = True
+        diagnostic_object = _create_diagnostic_entry().to_api_repr()
+        logging.info(diagnostic_object)
diff --git a/google/cloud/logging_v2/logger.py b/google/cloud/logging_v2/logger.py
index 02ecb6905..fa0af170c 100644
--- a/google/cloud/logging_v2/logger.py
+++ b/google/cloud/logging_v2/logger.py
@@ -23,6 +23,7 @@
 from google.cloud.logging_v2.entries import TextEntry
 from google.cloud.logging_v2.resource import Resource
 from google.cloud.logging_v2.handlers._monitored_resources import detect_resource
+from google.cloud.logging_v2._instrumentation import _add_instrumentation
 
 import google.protobuf.message
 
@@ -134,6 +135,7 @@ def _do_log(self, client, _entry_class, payload=None, **kw):
         kw["log_name"] = kw.pop("log_name", self.full_name)
         kw["labels"] = kw.pop("labels", self.labels)
         kw["resource"] = kw.pop("resource", self.default_resource)
+        partial_success = False
 
         severity = kw.get("severity", None)
         if isinstance(severity, str) and not severity.isupper():
@@ -155,7 +157,13 @@ def _do_log(self, client, _entry_class, payload=None, **kw):
             entry = _entry_class(**kw)
 
         api_repr = entry.to_api_repr()
-        client.logging_api.write_entries([api_repr])
+        entries = [api_repr]
+        if google.cloud.logging_v2._instrumentation_emitted is False:
+            partial_success = True
+            entries = _add_instrumentation(entries, **kw)
+            google.cloud.logging_v2._instrumentation_emitted = True
+
+        client.logging_api.write_entries(entries, partial_success=partial_success)
 
     def log_empty(self, *, client=None, **kw):
         """Log an empty message
diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index 84d0c9552..8d39408d3 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -888,6 +888,11 @@ def test_update_sink(self):
 
     @skip_for_mtls
     def test_api_equality_list_logs(self):
+        import google.cloud.logging_v2
+
+        # Skip diagnostic log for this system test
+        google.cloud.logging_v2._instrumentation_emitted = True
+
         unique_id = uuid.uuid1()
         gapic_logger = Config.CLIENT.logger(f"api-list-{unique_id}")
         http_logger = Config.HTTP_CLIENT.logger(f"api-list-{unique_id}")
diff --git a/tests/unit/handlers/test_structured_log.py b/tests/unit/handlers/test_structured_log.py
index 5031748f9..d2d570e21 100644
--- a/tests/unit/handlers/test_structured_log.py
+++ b/tests/unit/handlers/test_structured_log.py
@@ -438,3 +438,27 @@ def test_format_with_json_fields(self):
         self.assertEqual(result["message"], expected_result)
         self.assertEqual(result["hello"], "world")
         self.assertEqual(result["number"], 12)
+
+    def test_emits_instrumentation_info(self):
+        import logging
+        import mock
+        import google.cloud.logging_v2
+
+        handler = self._make_one()
+        logname = "loggername"
+        message = "Hello world!"
+
+        record = logging.LogRecord(logname, logging.INFO, "", 0, message, None, None)
+
+        with mock.patch.object(handler, "emit_instrumentation_info") as emit_info:
+
+            def side_effect():
+                google.cloud.logging_v2._instrumentation_emitted = True
+
+            emit_info.side_effect = side_effect
+            google.cloud.logging_v2._instrumentation_emitted = False
+            handler.emit(record)
+            handler.emit(record)
+
+            # emit_instrumentation_info should be called once
+            emit_info.assert_called_once()
diff --git a/tests/unit/test__instrumentation.py b/tests/unit/test__instrumentation.py
new file mode 100644
index 000000000..501301c34
--- /dev/null
+++ b/tests/unit/test__instrumentation.py
@@ -0,0 +1,65 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+import google.cloud.logging_v2._instrumentation as i
+
+
+class TestInstrumentation(unittest.TestCase):
+
+    TEST_NAME = "python"
+    # LONG_NAME > 14 characters
+    LONG_NAME = TEST_NAME + "789ABCDEF"
+
+    TEST_VERSION = "1.0.0"
+    # LONG_VERSION > 16 characters
+    LONG_VERSION = TEST_VERSION + "6789ABCDEF12"
+
+    def _get_diagonstic_value(self, entry, key):
+        return entry.payload[i._DIAGNOSTIC_INFO_KEY][i._INSTRUMENTATION_SOURCE_KEY][-1][
+            key
+        ]
+
+    def test_default_diagnostic_info(self):
+        entry = i._create_diagnostic_entry()
+        self.assertEqual(
+            i._PYTHON_LIBRARY_NAME,
+            self._get_diagonstic_value(entry, "name"),
+        )
+        self.assertEqual(
+            i._LIBRARY_VERSION, self._get_diagonstic_value(entry, "version")
+        )
+
+    def test_custom_diagnostic_info(self):
+        entry = i._create_diagnostic_entry(
+            name=self.TEST_NAME, version=self.TEST_VERSION
+        )
+        self.assertEqual(
+            self.TEST_NAME,
+            self._get_diagonstic_value(entry, "name"),
+        )
+        self.assertEqual(
+            self.TEST_VERSION, self._get_diagonstic_value(entry, "version")
+        )
+
+    def test_truncate_long_values(self):
+        entry = i._create_diagnostic_entry(
+            name=self.LONG_NAME, version=self.LONG_VERSION
+        )
+
+        expected_name = self.LONG_NAME[: i._MAX_NAME_LENGTH] + "*"
+        expected_version = self.LONG_VERSION[: i._MAX_VERSION_LENGTH] + "*"
+
+        self.assertEqual(expected_name, self._get_diagonstic_value(entry, "name"))
+        self.assertEqual(expected_version, self._get_diagonstic_value(entry, "version"))
diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py
index 597313824..a5d01898b 100644
--- a/tests/unit/test_logger.py
+++ b/tests/unit/test_logger.py
@@ -34,6 +34,12 @@ class TestLogger(unittest.TestCase):
     LOGGER_NAME = "logger-name"
     TIME_FORMAT = '"%Y-%m-%dT%H:%M:%S.%f%z"'
 
+    def setUp(self):
+        import google.cloud.logging_v2
+
+        # Test instrumentation behavior in only one test
+        google.cloud.logging_v2._instrumentation_emitted = True
+
     @staticmethod
     def _get_target_class():
         from google.cloud.logging import Logger
@@ -975,6 +981,43 @@ def test_list_entries_folder(self):
         self.assertIsNone(entry.logger)
         self.assertEqual(entry.log_name, LOG_NAME)
 
+    def test_first_log_emits_instrumentation(self):
+        from google.cloud.logging_v2.handlers._monitored_resources import (
+            detect_resource,
+        )
+        from google.cloud.logging_v2._instrumentation import _create_diagnostic_entry
+        import google.cloud.logging_v2
+
+        google.cloud.logging_v2._instrumentation_emitted = False
+        DEFAULT_LABELS = {"foo": "spam"}
+        resource = detect_resource(self.PROJECT)
+        instrumentation_entry = _create_diagnostic_entry(
+            resource=resource,
+            labels=DEFAULT_LABELS,
+        ).to_api_repr()
+        instrumentation_entry["logName"] = "projects/%s/logs/%s" % (
+            self.PROJECT,
+            self.LOGGER_NAME,
+        )
+        ENTRIES = [
+            instrumentation_entry,
+            {
+                "logName": "projects/%s/logs/%s" % (self.PROJECT, self.LOGGER_NAME),
+                "resource": resource._to_dict(),
+                "labels": DEFAULT_LABELS,
+            },
+        ]
+        client = _Client(self.PROJECT)
+        api = client.logging_api = _DummyLoggingAPI()
+        logger = self._make_one(self.LOGGER_NAME, client=client, labels=DEFAULT_LABELS)
+        logger.log_empty()
+        self.assertEqual(api._write_entries_called_with, (ENTRIES, None, None, None))
+
+        ENTRIES = ENTRIES[-1:]
+        api = client.logging_api = _DummyLoggingAPI()
+        logger.log_empty()
+        self.assertEqual(api._write_entries_called_with, (ENTRIES, None, None, None))
+
 
 class TestBatch(unittest.TestCase):
 
@@ -1645,7 +1688,15 @@ class _DummyLoggingAPI(object):
 
     _write_entries_called_with = None
 
-    def write_entries(self, entries, *, logger_name=None, resource=None, labels=None):
+    def write_entries(
+        self,
+        entries,
+        *,
+        logger_name=None,
+        resource=None,
+        labels=None,
+        partial_success=False,
+    ):
         self._write_entries_called_with = (entries, logger_name, resource, labels)
 
     def logger_delete(self, logger_name):