7219519135
This patch set is meant to create a workflow that will allow us to upgrade the airflow worker without causing disruption to the current running workflow. Note that we will set the update strategy for airflow worker to 'OnDelete'. The 'OnDelete' update strategy implements the legacy (1.6 and prior) behavior. When we select this update strategy, the statefulSet controller will not automatically update Pods when a modification is made to the StatefulSet’s '.spec.template field'. This strategy can be selected by setting the '.spec.template.updateStrategy.type' to 'OnDelete'. Refer to [0] for more information. [0] https://kubernetes.io/docs/tutorials/stateful-application/basic-stateful-set/#creating-a-statefulset Change-Id: I1f6c3564b7fba6abe422b86e36818eb2cd3454ea
43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
# Copyright 2018 AT&T Intellectual Property. All other rights reserved.
|
|
#
|
|
# 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 logging
|
|
|
|
from airflow.exceptions import AirflowException
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class XcomPusher(object):
|
|
"""XcomPusher pushes xcom value
|
|
|
|
Create specific key with value and stores as xcom.
|
|
"""
|
|
|
|
def __init__(self, task_instance):
|
|
self.ti = task_instance
|
|
|
|
def xcom_push(self, key=None, value=None):
|
|
"""Push a particular xcom value"""
|
|
|
|
LOG.info("Pushing xcom from %s.%s with key %s and value %s",
|
|
self.ti.dag_id,
|
|
self.ti.task_id,
|
|
key,
|
|
value)
|
|
|
|
try:
|
|
self.ti.xcom_push(key=key, value=value)
|
|
except:
|
|
raise AirflowException("Xcom push failed!")
|