You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by jl...@apache.org on 2017/02/18 16:12:24 UTC
incubator-airflow git commit: [AIRFLOW-871] change logging.warn()
into warning()
Repository: incubator-airflow
Updated Branches:
refs/heads/master baa4cd680 -> 21d775a9a
[AIRFLOW-871] change logging.warn() into warning()
This silences deprecation warnings, e.g.
airflow/airflow/utils/dag_processing.py:578:
DeprecationWarning: The
'warn' method is deprecated, use 'warning' instead
Closes #2082 from imbaczek/bug871
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/21d775a9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/21d775a9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/21d775a9
Branch: refs/heads/master
Commit: 21d775a9a4dc10cba7a73e373b2262ac9e3994b6
Parents: baa4cd6
Author: Marek Baczynski <ma...@intel.com>
Authored: Sat Feb 18 11:12:14 2017 -0500
Committer: Jeremiah Lowin <jl...@apache.org>
Committed: Sat Feb 18 11:12:14 2017 -0500
----------------------------------------------------------------------
airflow/api/auth/backend/kerberos_auth.py | 2 +-
airflow/contrib/auth/backends/ldap_auth.py | 4 +-
airflow/contrib/executors/mesos_executor.py | 2 +-
.../contrib/task_runner/cgroup_task_runner.py | 14 ++---
airflow/jobs.py | 40 ++++++-------
airflow/models.py | 2 +-
airflow/utils/dag_processing.py | 19 +++---
airflow/utils/helpers.py | 62 ++++++++++----------
setup.py | 4 +-
9 files changed, 75 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/api/auth/backend/kerberos_auth.py
----------------------------------------------------------------------
diff --git a/airflow/api/auth/backend/kerberos_auth.py b/airflow/api/auth/backend/kerberos_auth.py
index 7fdef12..d1c3b70 100644
--- a/airflow/api/auth/backend/kerberos_auth.py
+++ b/airflow/api/auth/backend/kerberos_auth.py
@@ -65,7 +65,7 @@ def init_app(app):
logging.info("Kerberos init: {} {}".format(service, hostname))
principal = kerberos.getServerPrincipalDetails(service, hostname)
except kerberos.KrbError as err:
- logging.warn("Kerberos: {}".format(err))
+ logging.warning("Kerberos: {}".format(err))
else:
logging.info("Kerberos API: server is {}".format(principal))
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/contrib/auth/backends/ldap_auth.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/auth/backends/ldap_auth.py b/airflow/contrib/auth/backends/ldap_auth.py
index 24a63bc..65af300 100644
--- a/airflow/contrib/auth/backends/ldap_auth.py
+++ b/airflow/contrib/auth/backends/ldap_auth.py
@@ -72,7 +72,7 @@ def get_ldap_connection(dn=None, password=None):
def group_contains_user(conn, search_base, group_filter, user_name_attr, username):
search_filter = '(&({0}))'.format(group_filter)
if not conn.search(search_base, search_filter, attributes=[user_name_attr]):
- LOG.warn("Unable to find group for %s %s", search_base, search_filter)
+ LOG.warning("Unable to find group for %s %s", search_base, search_filter)
else:
for resp in conn.response:
if (
@@ -93,7 +93,7 @@ def groups_user(conn, search_base, user_filter, user_name_att, username):
raise AuthenticationError("Invalid username or password")
if conn.response and "memberOf" not in conn.response[0]["attributes"]:
- LOG.warn("""Missing attribute "memberOf" when looked-up in Ldap database.
+ LOG.warning("""Missing attribute "memberOf" when looked-up in Ldap database.
The user does not seem to be a member of a group and therefore won't see any dag
if the option filter_by_owner=True and owner_mode=ldapgroup are set""")
return []
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/contrib/executors/mesos_executor.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/executors/mesos_executor.py b/airflow/contrib/executors/mesos_executor.py
index f2628a4..49788fc 100644
--- a/airflow/contrib/executors/mesos_executor.py
+++ b/airflow/contrib/executors/mesos_executor.py
@@ -167,7 +167,7 @@ class AirflowMesosScheduler(mesos.interface.Scheduler):
except KeyError:
# The map may not contain an item if the framework re-registered after a failover.
# Discard these tasks.
- logging.warn("Unrecognised task key %s" % update.task_id.value)
+ logging.warning("Unrecognised task key %s" % update.task_id.value)
return
if update.state == mesos_pb2.TASK_FINISHED:
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/contrib/task_runner/cgroup_task_runner.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/task_runner/cgroup_task_runner.py b/airflow/contrib/task_runner/cgroup_task_runner.py
index 6a9e6cf..11c45c1 100644
--- a/airflow/contrib/task_runner/cgroup_task_runner.py
+++ b/airflow/contrib/task_runner/cgroup_task_runner.py
@@ -95,8 +95,8 @@ class CgroupTaskRunner(BaseTaskRunner):
for path_element in path_split:
name_to_node = {x.name: x for x in node.children}
if path_element not in name_to_node:
- self.logger.warn("Cgroup does not exist: {}"
- .format(path))
+ self.logger.warning("Cgroup does not exist: {}"
+ .format(path))
return
else:
node = name_to_node[path_element]
@@ -164,11 +164,11 @@ class CgroupTaskRunner(BaseTaskRunner):
# I wasn't able to track down the root cause of the package install failures, but
# we might want to revisit that approach at some other point.
if return_code == 137:
- self.logger.warn("Task failed with return code of 137. This may indicate "
- "that it was killed due to excessive memory usage. "
- "Please consider optimizing your task or using the "
- "resources argument to reserve more memory for your "
- "task")
+ self.logger.warning("Task failed with return code of 137. This may indicate "
+ "that it was killed due to excessive memory usage. "
+ "Please consider optimizing your task or using the "
+ "resources argument to reserve more memory for your "
+ "task")
return return_code
def terminate(self):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/jobs.py
----------------------------------------------------------------------
diff --git a/airflow/jobs.py b/airflow/jobs.py
index fedad55..f385a6b 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -369,7 +369,7 @@ class DagFileProcessor(AbstractDagFileProcessor):
# Arbitrarily wait 5s for the process to die
self._process.join(5)
if sigkill and self._process.is_alive():
- logging.warn("Killing PID %s", self._process.pid)
+ logging.warning("Killing PID %s", self._process.pid)
os.kill(self._process.pid, signal.SIGKILL)
@property
@@ -541,7 +541,7 @@ class SchedulerJob(BaseJob):
"""
if not any([ti.sla for ti in dag.tasks]):
self.logger.info("Skipping SLA check for {} because "
- "no tasks in DAG have SLAs".format(dag))
+ "no tasks in DAG have SLAs".format(dag))
return
TI = models.TaskInstance
@@ -923,25 +923,25 @@ class SchedulerJob(BaseJob):
)
if len(dag_runs) == 0:
- self.logger.warn("DagRun for %s %s does not exist",
- task_instance.dag_id,
- task_instance.execution_date)
+ self.logger.warning("DagRun for %s %s does not exist",
+ task_instance.dag_id,
+ task_instance.execution_date)
continue
# There should only be one DAG run. Add some logging info if this
# is not the case for later debugging.
if len(dag_runs) > 1:
- self.logger.warn("Multiple DagRuns found for {} {}: {}"
- .format(task_instance.dag_id,
- task_instance.execution_date,
- dag_runs))
+ self.logger.warning("Multiple DagRuns found for {} {}: {}"
+ .format(task_instance.dag_id,
+ task_instance.execution_date,
+ dag_runs))
if not any(dag_run.state == State.RUNNING for dag_run in dag_runs):
- self.logger.warn("Setting {} to state={} as it does not have "
- "a DagRun in the {} state"
- .format(task_instance,
- new_state,
- State.RUNNING))
+ self.logger.warning("Setting {} to state={} as it does not have "
+ "a DagRun in the {} state"
+ .format(task_instance,
+ new_state,
+ State.RUNNING))
task_instance.state = new_state
session.merge(task_instance)
session.commit()
@@ -1534,7 +1534,7 @@ class SchedulerJob(BaseJob):
.format(dagbag.dags.keys(),
file_path))
else:
- self.logger.warn("No viable dags retrieved from {}".format(file_path))
+ self.logger.warning("No viable dags retrieved from {}".format(file_path))
self.update_import_errors(session, dagbag)
return []
@@ -1836,17 +1836,17 @@ class BackfillJob(BaseJob):
# If the set of tasks that aren't ready ever equals the set of
# tasks to run, then the backfill is deadlocked
if not_ready and not_ready == set(tasks_to_run):
- self.logger.warn("Deadlock discovered for tasks_to_run={}"
- .format(tasks_to_run.values()))
+ self.logger.warning("Deadlock discovered for tasks_to_run={}"
+ .format(tasks_to_run.values()))
deadlocked.update(tasks_to_run.values())
tasks_to_run.clear()
# Reacting to events
for key, state in list(executor.get_event_buffer().items()):
if key not in tasks_to_run:
- self.logger.warn("{} state {} not in tasks_to_run={}"
- .format(key, state,
- tasks_to_run.values()))
+ self.logger.warning("{} state {} not in tasks_to_run={}"
+ .format(key, state,
+ tasks_to_run.values()))
continue
ti = tasks_to_run[key]
ti.refresh_from_db()
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 8ac8c72..a6c100e 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -1308,7 +1308,7 @@ class TaskInstance(Base):
# the current worker process was blocked on refresh_from_db
if self.state == State.RUNNING:
msg = "Task Instance already running {}".format(self)
- logging.warn(msg)
+ logging.warning(msg)
session.commit()
return
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/utils/dag_processing.py
----------------------------------------------------------------------
diff --git a/airflow/utils/dag_processing.py b/airflow/utils/dag_processing.py
index 8cf9c0f..cdcc7b0 100644
--- a/airflow/utils/dag_processing.py
+++ b/airflow/utils/dag_processing.py
@@ -307,6 +307,7 @@ class DagFileProcessorManager(LoggingMixin):
:type _last_runtime: dict[unicode, float]
:type _last_finish_time: dict[unicode, datetime]
"""
+
def __init__(self,
dag_directory,
file_paths,
@@ -442,7 +443,7 @@ class DagFileProcessorManager(LoggingMixin):
if file_path in new_file_paths:
filtered_processors[file_path] = processor
else:
- self.logger.warn("Stopping processor for {}".format(file_path))
+ self.logger.warning("Stopping processor for {}".format(file_path))
processor.stop()
self._processors = filtered_processors
@@ -477,7 +478,7 @@ class DagFileProcessorManager(LoggingMixin):
"""
now = datetime.now()
return os.path.join(self._child_process_log_directory,
- now.strftime("%Y-%m-%d"))
+ now.strftime("%Y-%m-%d"))
def _get_log_file_path(self, dag_file_path):
"""
@@ -518,8 +519,8 @@ class DagFileProcessorManager(LoggingMixin):
os.symlink(log_directory, latest_log_directory_path)
elif (os.path.isdir(latest_log_directory_path) or
os.path.isfile(latest_log_directory_path)):
- self.logger.warn("{} already exists as a dir/file. "
- "Skip creating symlink."
+ self.logger.warning("{} already exists as a dir/file. "
+ "Skip creating symlink."
.format(latest_log_directory_path))
else:
os.symlink(log_directory, latest_log_directory_path)
@@ -571,11 +572,11 @@ class DagFileProcessorManager(LoggingMixin):
simple_dags = []
for file_path, processor in finished_processors.items():
if processor.result is None:
- self.logger.warn("Processor for {} exited with return code "
- "{}. See {} for details."
- .format(processor.file_path,
- processor.exit_code,
- processor.log_file))
+ self.logger.warning("Processor for {} exited with return code "
+ "{}. See {} for details."
+ .format(processor.file_path,
+ processor.exit_code,
+ processor.log_file))
else:
for simple_dag in processor.result:
simple_dags.append(simple_dag)
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/airflow/utils/helpers.py
----------------------------------------------------------------------
diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index e66745c..dee0657 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -206,7 +206,7 @@ def kill_process_tree(logger, pid):
try:
root_process = psutil.Process(pid)
except psutil.NoSuchProcess:
- logger.warn("PID: {} does not exist".format(pid))
+ logger.warning("PID: {} does not exist".format(pid))
return
# Check child processes to reduce cases where a child process died but
@@ -215,26 +215,26 @@ def kill_process_tree(logger, pid):
if x.is_running()]
if len(descendant_processes) != 0:
- logger.warn("Terminating descendant processes of {} PID: {}"
- .format(root_process.cmdline(),
- root_process.pid))
+ logger.warning("Terminating descendant processes of {} PID: {}"
+ .format(root_process.cmdline(),
+ root_process.pid))
temp_processes = descendant_processes[:]
for descendant in temp_processes:
- logger.warn("Terminating descendant process {} PID: {}"
- .format(descendant.cmdline(), descendant.pid))
+ logger.warning("Terminating descendant process {} PID: {}"
+ .format(descendant.cmdline(), descendant.pid))
try:
kill_using_shell(descendant.pid, signal.SIGTERM)
except psutil.NoSuchProcess:
descendant_processes.remove(descendant)
- logger.warn("Waiting up to {}s for processes to exit..."
- .format(TIME_TO_WAIT_AFTER_SIGTERM))
+ logger.warning("Waiting up to {}s for processes to exit..."
+ .format(TIME_TO_WAIT_AFTER_SIGTERM))
try:
psutil.wait_procs(descendant_processes, TIME_TO_WAIT_AFTER_SIGTERM)
- logger.warn("Done waiting")
+ logger.warning("Done waiting")
except psutil.TimeoutExpired:
- logger.warn("Ran out of time while waiting for "
- "processes to exit")
+ logger.warning("Ran out of time while waiting for "
+ "processes to exit")
# Then SIGKILL
descendant_processes = [x for x in root_process.children(recursive=True)
if x.is_running()]
@@ -242,16 +242,16 @@ def kill_process_tree(logger, pid):
if len(descendant_processes) > 0:
temp_processes = descendant_processes[:]
for descendant in temp_processes:
- logger.warn("Killing descendant process {} PID: {}"
- .format(descendant.cmdline(), descendant.pid))
+ logger.warning("Killing descendant process {} PID: {}"
+ .format(descendant.cmdline(), descendant.pid))
try:
kill_using_shell(descendant.pid, signal.SIGTERM)
descendant.wait()
except psutil.NoSuchProcess:
descendant_processes.remove(descendant)
- logger.warn("Killed all descendant processes of {} PID: {}"
- .format(root_process.cmdline(),
- root_process.pid))
+ logger.warning("Killed all descendant processes of {} PID: {}"
+ .format(root_process.cmdline(),
+ root_process.pid))
else:
logger.debug("There are no descendant processes to kill")
@@ -279,27 +279,27 @@ def kill_descendant_processes(logger, pids_to_kill=None):
if len(descendant_processes) == 0:
logger.debug("There are no descendant processes that can be killed")
return
- logger.warn("Terminating descendant processes of {} PID: {}"
- .format(this_process.cmdline(),
- this_process.pid))
+ logger.warning("Terminating descendant processes of {} PID: {}"
+ .format(this_process.cmdline(),
+ this_process.pid))
temp_processes = descendant_processes[:]
for descendant in temp_processes:
try:
- logger.warn("Terminating descendant process {} PID: {}"
- .format(descendant.cmdline(), descendant.pid))
+ logger.warning("Terminating descendant process {} PID: {}"
+ .format(descendant.cmdline(), descendant.pid))
descendant.terminate()
except psutil.NoSuchProcess:
descendant_processes.remove(descendant)
- logger.warn("Waiting up to {}s for processes to exit..."
- .format(TIME_TO_WAIT_AFTER_SIGTERM))
+ logger.warning("Waiting up to {}s for processes to exit..."
+ .format(TIME_TO_WAIT_AFTER_SIGTERM))
try:
psutil.wait_procs(descendant_processes, TIME_TO_WAIT_AFTER_SIGTERM)
- logger.warn("Done waiting")
+ logger.warning("Done waiting")
except psutil.TimeoutExpired:
- logger.warn("Ran out of time while waiting for "
- "processes to exit")
+ logger.warning("Ran out of time while waiting for "
+ "processes to exit")
# Then SIGKILL
descendant_processes = [x for x in this_process.children(recursive=True)
if x.is_running()]
@@ -309,16 +309,16 @@ def kill_descendant_processes(logger, pids_to_kill=None):
if len(descendant_processes) > 0:
for descendant in descendant_processes:
- logger.warn("Killing descendant process {} PID: {}"
- .format(descendant.cmdline(), descendant.pid))
+ logger.warning("Killing descendant process {} PID: {}"
+ .format(descendant.cmdline(), descendant.pid))
try:
descendant.kill()
descendant.wait()
except psutil.NoSuchProcess:
pass
- logger.warn("Killed all descendant processes of {} PID: {}"
- .format(this_process.cmdline(),
- this_process.pid))
+ logger.warning("Killed all descendant processes of {} PID: {}"
+ .format(this_process.cmdline(),
+ this_process.pid))
class AirflowImporter(object):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/21d775a9/setup.py
----------------------------------------------------------------------
diff --git a/setup.py b/setup.py
index a35e71b..880852a 100644
--- a/setup.py
+++ b/setup.py
@@ -68,10 +68,10 @@ def git_version(version):
import git
repo = git.Repo('.git')
except ImportError:
- logger.warn('gitpython not found: Cannot compute the git version.')
+ logger.warning('gitpython not found: Cannot compute the git version.')
return ''
except Exception as e:
- logger.warn('Git repo not found: Cannot compute the git version.')
+ logger.warning('Git repo not found: Cannot compute the git version.')
return ''
if repo:
sha = repo.head.commit.hexsha