[ https://issues.apache.org/jira/browse/AIRFLOW-5447?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16930761#comment-16930761 ]
Daniel Imberman edited comment on AIRFLOW-5447 at 9/16/19 6:32 PM: ------------------------------------------------------------------- Ok so I've broken down the currently running threads in hopes this helps us out Thread 1: attempting to put a new task in the task_queue {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1439, in _execute_helper self.executor.heartbeat() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 132, in heartbeat self.trigger_tasks(open_slots) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 156, in trigger_tasks executor_config=simple_ti.executor_config) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 767, in execute_async self.task_queue.put((key, command, kube_executor_config)) File "<string>", line 2, in put File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 819, in _callmethod kind, result = conn.recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 250, in recv buf = self._recv_bytes() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes buf = self._recv(4) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 379, in _recv chunk = read(handle, remaining) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 2: re-reading plugins files {code:java} Thread 0x7f0c01c31700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs)Thread 0x7f0bff430700 File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 201, in handle_request result = func(c, *args, **kwds) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 422, in accept_connection self.serve_client(c) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 234, in serve_client request = recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 251, in recv return _ForkingPickler.loads(buf.getbuffer()) File "<frozen importlib._bootstrap>", line 202, in _lock_unlock_module File "<frozen importlib._bootstrap>", line 98, in acquire File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 178, in accepter c = self.listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 453, in accept c = self._listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 598, in accept s, self._last_accepted = self._socket.accept() File "/usr/local/lib/python3.7/socket.py", line 212, in accept fd, addr = self._accept() File "/usr/local/airflow/.local/bin/airflow", line 21, in <module> from airflow import configuration File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/__init__.py", line 94, in <module> operators._integrate_plugins() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/operators/__init__.py", line 104, in _integrate_plugins from airflow.plugins_manager import operators_modules File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked {code} Thread 3: thread manager server {code:java} Thread 0x7f0c13c7d700 File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/plugins_manager.py", line 142, in <module> m = imp.load_source(namespace, filepath) File "/usr/local/lib/python3.7/imp.py", line 171, in load_source module = _load(spec) File "<frozen importlib._bootstrap>", line 696, in _load File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/plugins/parameterized_dags/__init__.py", line 4, in <module> from airflow.www.app import csrf File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/www/app.py", line 37, in <module> from airflow.www.blueprints import routes File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/www/blueprints.py", line 24, in <module> from airflow import jobs File "<frozen importlib._bootstrap>", line 1035, in _handle_fromlist File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/__init__.py", line 21, in <module> from airflow.jobs.base_job import BaseJob # noqa: F401 File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 48, in <module> class BaseJob(Base, LoggingMixin): File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 82, in BaseJob executor=executors.get_default_executor(), File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/__init__.py", line 48, in get_default_executor DEFAULT_EXECUTOR = _get_executor(executor_name) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/__init__.py", line 86, in _get_executor return KubernetesExecutor() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 650, in __init__ self._manager = multiprocessing.Manager() File "/usr/local/lib/python3.7/multiprocessing/context.py", line 56, in Manager m.start() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 563, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 597, in _run_server server.serve_forever() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 165, in serve_forever self.stop_event.wait(1) File "/usr/local/lib/python3.7/threading.py", line 552, in wait signaled = self._cond.wait(timeout) File "/usr/local/lib/python3.7/threading.py", line 300, in wait gotit = waiter.acquire(True, timeout) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 4: a second(?) threadmanager server {code:java} Thread 0x7f0c01c31700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 201, in handle_request result = func(c, *args, **kwds) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 422, in accept_connection self.serve_client(c) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 234, in serve_client request = recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 250, in recv buf = self._recv_bytes() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes buf = self._recv(4) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 379, in _recv chunk = read(handle, remaining)Thread 0x7f0bff430700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 178, in accepter c = self.listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 453, in accept c = self._listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 598, in accept s, self._last_accepted = self._socket.accept() File "/usr/local/lib/python3.7/socket.py", line 212, in accept fd, addr = self._accept()Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1374, in _execute_helper self.executor.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 756, in start self.kube_client, self.worker_uuid File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 412, in __init__ self._manager = multiprocessing.Manager() File "/usr/local/lib/python3.7/multiprocessing/context.py", line 56, in Manager m.start() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 563, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 597, in _run_server server.serve_forever() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 165, in serve_forever self.stop_event.wait(1) File "/usr/local/lib/python3.7/threading.py", line 552, in wait signaled = self._cond.wait(timeout) File "/usr/local/lib/python3.7/threading.py", line 300, in wait gotit = waiter.acquire(True, timeout) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 5: Jobwatcher reading k8s stream {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1374, in _execute_helper self.executor.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 756, in start self.kube_client, self.worker_uuid File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 415, in __init__ self.kube_watcher = self._make_kube_watcher() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 421, in _make_kube_watcher watcher.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 223, in _Popen return _default_context.get_context().Process._Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 325, in run self.worker_uuid, self.kube_config) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 349, in _run **kwargs): File "/usr/local/airflow/.local/lib/python3.7/site-packages/kubernetes/watch/watch.py", line 144, in stream for line in iter_resp_lines(resp): File "/usr/local/airflow/.local/lib/python3.7/site-packages/kubernetes/watch/watch.py", line 48, in iter_resp_lines for seg in resp.read_chunked(decode_content=False): File "/usr/local/lib/python3.7/site-packages/urllib3/response.py", line 704, in read_chunked self._update_chunk_length() File "/usr/local/lib/python3.7/site-packages/urllib3/response.py", line 636, in _update_chunk_length line = self._fp.fp.readline() File "/usr/local/lib/python3.7/socket.py", line 589, in readinto return self._sock.recv_into(b) File "/usr/local/lib/python3.7/site-packages/urllib3/contrib/pyopenssl.py", line 304, in recv_into return self.connection.recv_into(*args, **kwargs) File "/usr/local/lib/python3.7/site-packages/OpenSSL/SSL.py", line 1821, in recv_into result = _lib.SSL_read(self._ssl, buf, nbytes) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 6: reading DAG files {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1380, in _execute_helper self.processor_agent.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 543, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 223, in _Popen return _default_context.get_context().Process._Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 613, in _run_processor_manager processor_manager.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 872, in start self._signal_conn.send(dag_parsing_stat) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 206, in send self._send_bytes(_ForkingPickler.dumps(obj)) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 404, in _send_bytes self._send(header + buf) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 368, in _send n = write(self._handle, buf) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} was (Author: dimberman): Ok so I've broken down the currently running threads in hopes this helps us out Thread 1: attempting to put a new task in the task_queue {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1439, in _execute_helper self.executor.heartbeat() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 132, in heartbeat self.trigger_tasks(open_slots) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 156, in trigger_tasks executor_config=simple_ti.executor_config) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 767, in execute_async self.task_queue.put((key, command, kube_executor_config)) File "<string>", line 2, in put File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 819, in _callmethod kind, result = conn.recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 250, in recv buf = self._recv_bytes() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes buf = self._recv(4) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 379, in _recv chunk = read(handle, remaining) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 2: re-reading plugins files {code:java} Thread 0x7f0c01c31700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs)Thread 0x7f0bff430700 File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 201, in handle_request result = func(c, *args, **kwds) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 422, in accept_connection self.serve_client(c) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 234, in serve_client request = recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 251, in recv return _ForkingPickler.loads(buf.getbuffer()) File "<frozen importlib._bootstrap>", line 202, in _lock_unlock_module File "<frozen importlib._bootstrap>", line 98, in acquire File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 178, in accepter c = self.listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 453, in accept c = self._listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 598, in accept s, self._last_accepted = self._socket.accept() File "/usr/local/lib/python3.7/socket.py", line 212, in accept fd, addr = self._accept() File "/usr/local/airflow/.local/bin/airflow", line 21, in <module> from airflow import configuration File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/__init__.py", line 94, in <module> operators._integrate_plugins() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/operators/__init__.py", line 104, in _integrate_plugins from airflow.plugins_manager import operators_modules File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked {code} Thread 3: thread manager server {code:java} Thread 0x7f0c13c7d700 File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/plugins_manager.py", line 142, in <module> m = imp.load_source(namespace, filepath) File "/usr/local/lib/python3.7/imp.py", line 171, in load_source module = _load(spec) File "<frozen importlib._bootstrap>", line 696, in _load File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/plugins/parameterized_dags/__init__.py", line 4, in <module> from airflow.www.app import csrf File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/www/app.py", line 37, in <module> from airflow.www.blueprints import routes File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/www/blueprints.py", line 24, in <module> from airflow import jobs File "<frozen importlib._bootstrap>", line 1035, in _handle_fromlist File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/__init__.py", line 21, in <module> from airflow.jobs.base_job import BaseJob # noqa: F401 File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 48, in <module> class BaseJob(Base, LoggingMixin): File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 82, in BaseJob executor=executors.get_default_executor(), File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/__init__.py", line 48, in get_default_executor DEFAULT_EXECUTOR = _get_executor(executor_name) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/executors/__init__.py", line 86, in _get_executor return KubernetesExecutor() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 650, in __init__ self._manager = multiprocessing.Manager() File "/usr/local/lib/python3.7/multiprocessing/context.py", line 56, in Manager m.start() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 563, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 597, in _run_server server.serve_forever() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 165, in serve_forever self.stop_event.wait(1) File "/usr/local/lib/python3.7/threading.py", line 552, in wait signaled = self._cond.wait(timeout) File "/usr/local/lib/python3.7/threading.py", line 300, in wait gotit = waiter.acquire(True, timeout) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 4: a second(?) threadmanager server {code:java} Thread 0x7f0c01c31700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 201, in handle_request result = func(c, *args, **kwds) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 422, in accept_connection self.serve_client(c) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 234, in serve_client request = recv() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 250, in recv buf = self._recv_bytes() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes buf = self._recv(4) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 379, in _recv chunk = read(handle, remaining)Thread 0x7f0bff430700 File "/usr/local/lib/python3.7/threading.py", line 890, in _bootstrap self._bootstrap_inner() File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner self.run() File "/usr/local/lib/python3.7/threading.py", line 870, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 178, in accepter c = self.listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 453, in accept c = self._listener.accept() File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 598, in accept s, self._last_accepted = self._socket.accept() File "/usr/local/lib/python3.7/socket.py", line 212, in accept fd, addr = self._accept()Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1374, in _execute_helper self.executor.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 756, in start self.kube_client, self.worker_uuid File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 412, in __init__ self._manager = multiprocessing.Manager() File "/usr/local/lib/python3.7/multiprocessing/context.py", line 56, in Manager m.start() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 563, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 597, in _run_server server.serve_forever() File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 165, in serve_forever self.stop_event.wait(1) File "/usr/local/lib/python3.7/threading.py", line 552, in wait signaled = self._cond.wait(timeout) File "/usr/local/lib/python3.7/threading.py", line 300, in wait gotit = waiter.acquire(True, timeout) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 5: Jobwatcher reading k8s stream {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1374, in _execute_helper self.executor.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 756, in start self.kube_client, self.worker_uuid File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 415, in __init__ self.kube_watcher = self._make_kube_watcher() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 421, in _make_kube_watcher watcher.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 223, in _Popen return _default_context.get_context().Process._Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 325, in run self.worker_uuid, self.kube_config) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 349, in _run **kwargs): File "/usr/local/airflow/.local/lib/python3.7/site-packages/kubernetes/watch/watch.py", line 144, in stream for line in iter_resp_lines(resp): File "/usr/local/airflow/.local/lib/python3.7/site-packages/kubernetes/watch/watch.py", line 48, in iter_resp_lines for seg in resp.read_chunked(decode_content=False): File "/usr/local/lib/python3.7/site-packages/urllib3/response.py", line 704, in read_chunked self._update_chunk_length() File "/usr/local/lib/python3.7/site-packages/urllib3/response.py", line 636, in _update_chunk_length line = self._fp.fp.readline() File "/usr/local/lib/python3.7/socket.py", line 589, in readinto return self._sock.recv_into(b) File "/usr/local/lib/python3.7/site-packages/urllib3/contrib/pyopenssl.py", line 304, in recv_into return self.connection.recv_into(*args, **kwargs) File "/usr/local/lib/python3.7/site-packages/OpenSSL/SSL.py", line 1821, in recv_into result = _lib.SSL_read(self._ssl, buf, nbytes) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} Thread 6: reading DAG files {code:java} Thread 0x7f0c13c7d700 File "/usr/local/airflow/.local/bin/airflow", line 32, in <module> args.func(args) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper return f(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/bin/cli.py", line 1013, in scheduler job.run() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 213, in run self._execute() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1350, in _execute self._execute_helper() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1380, in _execute_helper self.processor_agent.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 543, in start self._process.start() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 112, in start self._popen = self._Popen(self) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 223, in _Popen return _default_context.get_context().Process._Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/context.py", line 277, in _Popen return Popen(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 20, in __init__ self._launch(process_obj) File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 74, in _launch code = process_obj._bootstrap() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap self.run() File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run self._target(*self._args, **self._kwargs) File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 613, in _run_processor_manager processor_manager.start() File "/usr/local/airflow/.local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 872, in start self._signal_conn.send(dag_parsing_stat) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 206, in send self._send_bytes(_ForkingPickler.dumps(obj)) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 404, in _send_bytes self._send(header + buf) File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 368, in _send n = write(self._handle, buf) File "<string>", line 1, in <module> File "<string>", line 5, in <module> {code} > KubernetesExecutor hangs on task queueing > ----------------------------------------- > > Key: AIRFLOW-5447 > URL: https://issues.apache.org/jira/browse/AIRFLOW-5447 > Project: Apache Airflow > Issue Type: Bug > Components: executor-kubernetes > Affects Versions: 1.10.4, 1.10.5 > Environment: Kubernetes version v1.14.3, Airflow version 1.10.4-1.10.5 > Reporter: Henry Cohen > Assignee: Daniel Imberman > Priority: Blocker > > Starting in 1.10.4, and continuing in 1.10.5, when using the > KubernetesExecutor, with the webserver and scheduler running in the > kubernetes cluster, tasks are scheduled, but when added to the task queue, > the executor process hangs indefinitely. Based on log messages, it appears to > be stuck at this line > https://github.com/apache/airflow/blob/v1-10-stable/airflow/contrib/executors/kubernetes_executor.py#L761 -- This message was sent by Atlassian Jira (v8.3.2#803003)