)
# Define deployment spec
- spec = client.ExtensionsV1beta1DeploymentSpec(
+ spec = client.V1DeploymentSpec(
replicas=replicas,
+ selector=client.V1LabelSelector(match_labels=labels),
template=template
)
# Create deployment object
- deployment = client.ExtensionsV1beta1Deployment(
+ deployment = client.V1Deployment(
+ api_version="apps/v1",
kind="Deployment",
metadata=client.V1ObjectMeta(name=deployment_name),
spec=spec
_configure_api(location)
# Get deployment spec
- spec = client.ExtensionsV1beta1Api().read_namespaced_deployment(deployment, namespace)
+ spec = client.AppsV1Api().read_namespaced_deployment(deployment, namespace)
# Apply changes to spec
spec = modify(spec)
# Patch the deploy with updated spec
- client.ExtensionsV1beta1Api().patch_namespaced_deployment(deployment, namespace, spec)
+ client.AppsV1Api().patch_namespaced_deployment(deployment, namespace, spec)
def _execute_command_in_pod(location, namespace, pod_name, command):
'''
# Get API handles
_configure_api(kwargs.get("k8s_location"))
core = client.CoreV1Api()
- ext = client.ExtensionsV1beta1Api()
+ k8s_apps_v1_api_client = client.AppsV1Api()
# Parse the port mapping
container_ports, port_map = parse_ports(kwargs.get("ports", []))
dep = _create_deployment_object(component_name, containers, init_containers, replicas, volumes, labels, pull_secrets=k8sconfig["image_pull_secrets"])
# Have k8s deploy it
- ext.create_namespaced_deployment(namespace, dep)
+ k8s_apps_v1_api_client.create_namespaced_deployment(namespace, dep)
deployment_ok = True
deployment_description["deployment"] = _create_deployment_name(component_name)
core.delete_namespaced_service(_create_service_name(component_name), namespace)
# If the deployment was created but not the service, delete the deployment
if deployment_ok:
- client.ExtensionsV1beta1Api().delete_namespaced_deployment(_create_deployment_name(component_name), namespace, body=client.V1DeleteOptions())
+ client.AppsV1Api().delete_namespaced_deployment(_create_deployment_name(component_name), namespace, body=client.V1DeleteOptions())
raise e
return dep, deployment_description
# Have k8s delete the underlying pods and replicaset when deleting the deployment.
options = client.V1DeleteOptions(propagation_policy="Foreground")
- client.ExtensionsV1beta1Api().delete_namespaced_deployment(deployment_description["deployment"], namespace, body=options)
+ client.AppsV1Api().delete_namespaced_deployment(deployment_description["deployment"], namespace, body=options)
def is_available(location, namespace, component_name):
_configure_api(location)
- dep_status = client.AppsV1beta1Api().read_namespaced_deployment_status(_create_deployment_name(component_name), namespace)
+ dep_status = client.AppsV1Api().read_namespaced_deployment_status(_create_deployment_name(component_name), namespace)
# Check if the number of available replicas is equal to the number requested and that the replicas match the current spec
# This check can be used to verify completion of an initial deployment, a scale operation, or an update operation
return dep_status.status.available_replicas == dep_status.spec.replicas and dep_status.status.updated_replicas == dep_status.spec.replicas
client.AppsV1beta1DeploymentRollback(name=deployment, rollback_to=client.AppsV1beta1RollbackConfig(revision=rollback_to)))
# Read back the spec for the rolled-back deployment
- spec = client.ExtensionsV1beta1Api().read_namespaced_deployment(deployment, namespace)
+ spec = client.AppsV1Api().read_namespaced_deployment(deployment, namespace)
return spec.spec.template.spec.containers[0].image, spec.spec.replicas
def execute_command_in_deployment(deployment_description, command):
setup(
name='k8splugin',
description='Cloudify plugin for containerized components deployed using Kubernetes',
- version="3.1.0",
- author='J. F. Lucas, Michael Hwang, Tommy Carpenter, Joanna Jeremicz',
+ version="3.2.0",
+ author='J. F. Lucas, Michael Hwang, Tommy Carpenter, Joanna Jeremicz, Sylwia Jakubek',
packages=['k8splugin','k8sclient','configure'],
zip_safe=False,
install_requires=[
'python-consul>=0.6.0',
'onap-dcae-dcaepolicy-lib>=2.4.1',
- 'kubernetes==10.0.0',
+ 'kubernetes==11.0.0',
'cloudify-common>=5.0.0',
]
)
# org.onap.dcae
# ================================================================================
# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2020 Nokia. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# to get an API object
core = client.CoreV1Api()
ext = client.ExtensionsV1beta1Api()
+ appsv1 = client.AppsV1Api()
def pseudo_deploy(namespace, dep):
return dep
monkeypatch.setattr(ext,"create_namespaced_deployment", pseudo_deploy)
return ext
+ # patched_appsv1 returns an AppsV1Api object with the
+ # create_namespaced_deployment method stubbed out so that there
+ # is no attempt to call the k8s API server
+ def patched_appsv1():
+ monkeypatch.setattr(ext,"create_namespaced_deployment", pseudo_deploy)
+ return ext
+
def pseudo_configure(loc):
pass
monkeypatch.setattr(k8sclient.k8sclient,"_configure_api", pseudo_configure)
monkeypatch.setattr(client, "CoreV1Api", patched_core)
monkeypatch.setattr(client,"ExtensionsV1beta1Api", patched_ext)
+ monkeypatch.setattr(client,"AppsV1Api", patched_appsv1)