Add kernel capabilities in DockerOperator(#9142)

airflow/providers/docker/operators/docker.py

@@ -123,6 +123,8 @@ class DockerOperator(BaseOperator):
     :param tty: Allocate pseudo-TTY to the container
         This needs to be set see logs of the Docker container.
     :type tty: bool
+    :param cap_add: Include container capabilities
+    :type cap_add: list[str]
     """
     template_fields = ('command', 'environment', 'container_name')
     template_ext = ('.sh', '.bash',)
@@ -159,6 +161,7 @@ class DockerOperator(BaseOperator):
             auto_remove: bool = False,
             shm_size: Optional[int] = None,
             tty: Optional[bool] = False,
+            cap_add: Optional[Iterable[str]] = None,
             *args,
             **kwargs) -> None:
 
@@ -191,6 +194,7 @@ class DockerOperator(BaseOperator):
         self.docker_conn_id = docker_conn_id
         self.shm_size = shm_size
         self.tty = tty
+        self.cap_add = cap_add
         if kwargs.get('xcom_push') is not None:
             raise AirflowException("'xcom_push' was deprecated, use 'BaseOperator.do_xcom_push' instead")
 
@@ -231,7 +235,8 @@ class DockerOperator(BaseOperator):
                     dns=self.dns,
                     dns_search=self.dns_search,
                     cpu_shares=int(round(self.cpus * 1024)),
-                    mem_limit=self.mem_limit),
+                    mem_limit=self.mem_limit,
+                    cap_add=self.cap_add),
                 image=self.image,
                 user=self.user,
                 working_dir=self.working_dir,

tests/providers/docker/operators/test_docker.py

@@ -82,7 +82,8 @@ class TestDockerOperator(unittest.TestCase):
                                                                mem_limit=None,
                                                                auto_remove=False,
                                                                dns=None,
-                                                               dns_search=None)
+                                                               dns_search=None,
+                                                               cap_add=None)
         tempdir_mock.assert_called_once_with(dir='/host/airflow', prefix='airflowtmp')
         client_mock.images.assert_called_once_with(name='ubuntu:latest')
         client_mock.attach.assert_called_once_with(container='some_id', stdout=True,