diff --git a/changelogs/fragments/488-add-support-for-tail-logs.yaml b/changelogs/fragments/488-add-support-for-tail-logs.yaml
new file mode 100644
index 0000000000..a2b29f125c
--- /dev/null
+++ b/changelogs/fragments/488-add-support-for-tail-logs.yaml
@@ -0,0 +1,3 @@
+---
+minor_changes:
+- k8s_log - added the `tail_lines` parameter to limit the number of lines to be retrieved from the end of the logs (https://github.com/ansible-collections/kubernetes.core/issues/488).
diff --git a/plugins/modules/k8s_log.py b/plugins/modules/k8s_log.py
index 94e5d5f9e1..fd0c8eea35 100644
--- a/plugins/modules/k8s_log.py
+++ b/plugins/modules/k8s_log.py
@@ -67,6 +67,12 @@
     type: bool
     default: False
     version_added: '2.4.0'
+  tail_lines:
+    description:
+    - A number of lines from the end of the logs to retrieve.
+    required: no
+    type: int
+    version_added: '2.4.0'
 
 requirements:
   - "python >= 3.6"
@@ -106,6 +112,7 @@
     kind: DeploymentConfig
     namespace: testing
     name: example
+    tail_lines: 100
   register: log
 """
 
@@ -156,6 +163,7 @@ def argspec():
             since_seconds=dict(),
             label_selectors=dict(type="list", elements="str", default=[]),
             previous=dict(type="bool", default=False),
+            tail_lines=dict(type="int"),
         )
     )
     return args
@@ -204,6 +212,11 @@ def execute_module(svc, params):
     if params.get("previous"):
         kwargs.setdefault("query_params", {}).update({"previous": params["previous"]})
 
+    if params.get("tail_lines"):
+        kwargs.setdefault("query_params", {}).update(
+            {"tailLines": params["tail_lines"]}
+        )
+
     response = resource.log.get(
         name=name, namespace=namespace, serialize=False, **kwargs
     )
diff --git a/tests/integration/targets/k8s_log/tasks/main.yml b/tests/integration/targets/k8s_log/tasks/main.yml
index 299874f7fd..336e6e1c69 100644
--- a/tests/integration/targets/k8s_log/tasks/main.yml
+++ b/tests/integration/targets/k8s_log/tasks/main.yml
@@ -112,6 +112,46 @@
       assert:
         that: job_logs.log_lines[0] == "7"
 
+    - name: create a job that has 10 log lines
+      k8s:
+        state: present
+        wait: yes
+        wait_timeout: "{{ k8s_wait_timeout | default(omit) }}"
+        wait_condition:
+          type: Complete
+          status: 'True'
+        definition:
+          apiVersion: batch/v1
+          kind: Job
+          metadata:
+            name: multiline-log
+            namespace: "{{ test_namespace }}"
+          spec:
+            template:
+              spec:
+                containers:
+                  - name: busybox
+                    image: busybox
+                    command: ['sh']
+                    args: ['-c', 'for i in $(seq 0 9); do echo $i; done']
+                restartPolicy: Never
+            backoffLimit: 4
+
+    - name: retrieve last 5 log lines from the job
+      k8s_log:
+        api_version: batch/v1
+        kind: Job
+        namespace: "{{ test_namespace }}"
+        name: multiline-log
+        tail_lines: 5
+      register: tailed_log
+
+    # The log_lines by k8s_log always contain a trailing empty element,
+    # so if "tail"ing 5 lines, the length will be 6.
+    - name: verify that specific number of logs have been retrieved
+      assert:
+        that: tailed_log.log_lines | length == 5 + 1
+
   always:
     - name: ensure that namespace is removed
       k8s: