openshift_gcp.py

#!/usr/bin/env python

from datetime import datetime
import googleapiclient.discovery
import googleapiclient.errors
import jinja2
import json
import logging
import os
import re
import socket
import sys
import time
import traceback
import urllib2
import weakref
import yaml

inventory_class_name = 'OpenShiftGCP'

# The call to create a disk image has been seen to timeout with the default
# 60 second limit.
socket.setdefaulttimeout(120)

class OpenShiftGCP:
    def __init__(self, ocpinv):
        self.ocpinv = weakref.ref(ocpinv)
        self.cloudresourcemanagerAPI = googleapiclient.discovery.build('cloudresourcemanager', 'v1')
        self.computeAPI = googleapiclient.discovery.build('compute', 'v1')
        self.dnsAPI = googleapiclient.discovery.build('dns', 'v1')
        self.serviceusageAPI = googleapiclient.discovery.build('serviceusage', 'v1')
        self.set_dynamic_cloud_vars()

    def set_dynamic_cloud_vars(self):
        if not self.ocpinv().cluster_var('openshift_gcp_project'):
            self.set_openshift_gcp_project()
        if self.ocpinv().init_mode:
            return
        if self.ocpinv().cluster_var('demo_use_cloud_dns'):
            self.set_cluster_domain_dns_servers()
        if not self.ocpinv().cluster_var('openshift_master_cluster_hostname'):
            self.set_master_cluster_hostname_with_loadbalancer_ip()
        if self.ocpinv().cluster_var('demo_wildcard_dns'):
            self.set_wildcard_dns_vars()

    def set_openshift_gcp_project(self):
        if 'GOOGLE_APPLICATION_CREDENTIALS' in os.environ:
            self.set_openshift_gcp_project_from_google_creds()
        else:
            self.set_openshift_gcp_project_from_metadata()

    def set_openshift_gcp_project_from_google_creds(self):
        try:
            fh = open(os.environ.get('GOOGLE_APPLICATION_CREDENTIALS'))
            gcred = json.loads(fh.read())
            self.ocpinv().set_dynamic_cluster_var('openshift_gcp_project', gcred['project_id'])
        except:
            traceback.print_exec()
            raise Exception("Unable to determine openshift_gcp_project from GOOGLE_APPLICATION_CREDENTIALS")

    def set_openshift_gcp_project_from_metadata(self):
        try:
            url = 'http://metadata.google.internal/computeMetadata/v1/project/project-id'
            headers = { "Metadata-Flavor": "Google" }
            req = urllib2.Request(url, None, headers)
            resp = urllib2.urlopen(req)
            self.ocpinv().set_dynamic_cluster_var('openshift_gcp_project', resp.read())
        except:
            traceback.print_exc()
            raise Exception("Unable to determine openshift_gcp_project from GOOGLE_APPLICATION_CREDENTIALS")

    def set_cluster_domain_dns_servers(self):
        """
        Set demo_cluster_domain_dns_servers based on dynamically
        determined name server list.
        """
        cluster_zone = self.dnsAPI.managedZones().get(
            managedZone = self.ocpinv().cluster_var('demo_gcp_dns_zone_name'),
            project = self.ocpinv().cluster_var('openshift_gcp_project')
        ).execute()
        self.ocpinv().set_dynamic_cluster_var(
            'demo_cluster_domain_dns_servers',
            cluster_zone['nameServers']
        )

    def get_master_internal_loadbalancer_ip(self):
        gcp_prefix = self.ocpinv().cluster_var('openshift_gcp_prefix')
        resp = self.computeAPI.forwardingRules().list(
            filter = '(name="{}master")'.format(gcp_prefix),
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            region = self.ocpinv().cluster_var('openshift_gcp_region')
        ).execute()
        if resp.get('items', []):
            return resp['items'][0]['IPAddress']
        else:
            return ''

    def set_master_cluster_hostname_with_loadbalancer_ip(self):
        self.ocpinv().set_dynamic_cluster_var(
            'openshift_master_cluster_hostname',
            self.get_master_internal_loadbalancer_ip()
        )

    def set_wildcard_dns_vars(self):
        wildcard_dns = self.ocpinv().cluster_var('demo_wildcard_dns')
        master_ip, router_ip = self.get_master_and_router_ip()
        if master_ip:
            self.ocpinv().set_dynamic_cluster_var(
                'openshift_master_cluster_hostname',
                "master-internal.{}.{}".format(
                    self.get_master_internal_loadbalancer_ip(),
                    wildcard_dns
                )
            )
            self.ocpinv().set_dynamic_cluster_var(
                'openshift_master_cluster_public_hostname',
                "master.{}.{}".format(
                    master_ip,
                    wildcard_dns
                )
            )
        if router_ip:
            self.ocpinv().set_dynamic_cluster_var(
                'openshift_master_default_subdomain',
                "{}.{}".format(
                    router_ip,
                    wildcard_dns
                )
            )
        controller_ip, controller_public_ip = self.get_controller_ips()
        if controller_ip:
            self.ocpinv().set_dynamic_cluster_var(
                'demo_controller_hostname',
                "controller.{}.{}".format(
                    controller_ip,
                    wildcard_dns
                )
            )
            self.ocpinv().set_dynamic_cluster_var(
                'demo_controller_public_hostname',
                "controller.{}.{}".format(
                    controller_public_ip,
                    wildcard_dns
                )
            )

    def get_master_and_router_ip(self):
        gcp_prefix = self.ocpinv().cluster_var('openshift_gcp_prefix')
        if self.ocpinv().cluster_var('demo_shared_public_load_balancer'):
            ip = self.get_globaladdress_ip(gcp_prefix + 'public')
            return ip, ip
        else:
            master_ip = self.get_globaladdress_ip(gcp_prefix + 'master')
            router_ip = self.get_globaladdress_ip(gcp_prefix + 'router')
            return master_ip, router_ip

    def get_controller_ips(self):
        gcp_prefix = self.ocpinv().cluster_var('openshift_gcp_prefix')
        controller_hostname = gcp_prefix + 'controller'
        instance = self.get_instance_in_zone(
            controller_hostname,
            self.get_cluster_zones()[0]
        )
        if instance:
            primary_network_interface = instance['networkInterfaces'][0]
            return(
                 primary_network_interface['networkIP'],
                 primary_network_interface['accessConfigs'][0]['natIP']
            )
        else:
            return None, None

    def get_globaladdress_ip(self, name):
        resp = self.computeAPI.globalAddresses().list(
            filter = 'name="{}"'.format(name),
            project = self.ocpinv().cluster_var('openshift_gcp_project')
        ).execute()
        for address in resp.get('items', []):
            return address['address']

    def instance_fqdn(self, instance):
        zoneinfo = instance['zone'].rsplit('/', 3)
        project = zoneinfo[1]
        zonename = zoneinfo[3]
        return '%s.%s.c.%s.internal' % (
            instance['name'],
            zonename,
            project
        )

    def instance_belongs_to_cluster(self, instance):
        return(
            self.ocpinv().cluster_name == instance.get('labels',{}).get('openshift-cluster','')
        )

    def get_cluster_zones(self):
        """
        Return list of zones configured for the cluster or list of all zones
        for the region.
        """

        config_zones = self.ocpinv().cluster_var('demo_gcp_zones')
        if config_zones:
            return config_zones

        if not self.ocpinv().cluster_var('openshift_gcp_multizone'):
            gcp_zone = self.ocpinv().cluster_var('openshift_gcp_zone')
            self.ocpinv().cluster_config['demo_gcp_zones'] = [gcp_zone]
            return [gcp_zone]

        region = self.computeAPI.regions().get(
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            region = self.ocpinv().cluster_var('openshift_gcp_region')
        ).execute()
        gcp_zones = [
            zone_uri.rsplit('/',1)[-1] for zone_uri in region['zones']
        ]
        self.ocpinv().cluster_config['demo_gcp_zones'] = gcp_zones
        return gcp_zones

    def get_instance_in_zone(self, hostname, zone):
        try:
            instance = self.computeAPI.instances().get(
                instance = hostname.split('.')[0],
                project = self.ocpinv().cluster_var('openshift_gcp_project'),
                zone = zone
            ).execute()
        except googleapiclient.errors.HttpError:
            return None
        if self.instance_belongs_to_cluster(instance):
            return instance
        # False means found instance and instance does not belong to this cluster
        return False

    def get_instance(self, hostname):
        for zone in self.get_cluster_zones():
            instance = self.get_instance_in_zone(hostname, zone)
            if instance != None:
                return instance
        return None

    def get_cluster_instances(self):
        for zone in self.get_cluster_zones():
            for instance in self.get_cluster_instances_in_zone(zone):
                yield instance

    def get_cluster_instances_in_zone(self, zone):
        for instance in self.get_instances_in_zone(zone):
            if( self.instance_belongs_to_cluster(instance)
            and self.ansible_group_filter(instance) ):
                # FIXME - Should have a mechanism to exclude dynamic/autoscaling instances
                yield instance

    def get_instances_in_zone(self, zone):
        req = self.computeAPI.instances().list(
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            zone = zone
        )
        while req:
            resp = req.execute()
            for instance in resp.get('items', []):
                yield instance
            req = self.computeAPI.instances().list_next(
                previous_request = req,
                previous_response = resp
            )

    def instance_ansible_host_groups(self, instance):
        groups = []
        for item in instance['metadata']['items']:
            if( item['key'].startswith('ansible-host-group-')
            and item['value'] == 'true'):
                groups.append(item['key'][19:])
        if groups:
            return groups
        else:
            # Did not find any ansible-host-group-* metadata, default to nodes
            return ['nodes']

    def instance_openshift_node_group_name(self, instance):
        name = instance['labels'].get('openshift-node-group-name', 'compute')
        # Drop useless "node-config-" prefix if present
        if name.startswith('node-config-'):
            return name[12:]
        return name

    def instance_openshift_node_labels(self, instance):
        node_labels = {
            'failure-domain.beta.kubernetes.io/region':
                self.ocpinv().cluster_var('openshift_gcp_region'),
            'failure-domain.beta.kubernetes.io/zone':
                instance['zone'].rsplit('/',1)[1]
        }

        node_group_name = self.instance_openshift_node_group_name(instance)
        node_group_labels = self.ocpinv().cluster_config \
            .get('demo_openshift_node_groups', {}) \
            .get(node_group_name, {}) \
            .get('labels', {'node-role.kubernetes.io/'+node_group_name: 'true' })
        node_labels.update(node_group_labels)
        return node_labels

    def instance_add_ansible_vars(self, instance, hostvars):
        for item in instance['metadata']['items']:
            if item['key'].startswith('ansible-var-'):
                try:
                    value = json.loads(item['value'])
                except:
                    value = item['value']
                hostvars[item['key'][12:]] = value

    def instance_add_ip_vars(self, instance, hostvars):
        primary_network_interface = instance['networkInterfaces'][0]
        primary_network_ip = primary_network_interface['networkIP']
        hostvars['gcp_network_ip'] = primary_network_ip
        if primary_network_interface['accessConfigs'][0].get('natIP', None):
            nat_ip = primary_network_interface['accessConfigs'][0]['natIP']
            hostvars['gcp_nat_ip'] = nat_ip
            if os.environ.get('GCP_ANSIBLE_INVENTORY_USE_NAT_IP', 'false') == 'true':
                hostvars['ansible_host'] = nat_ip
            else:
                hostvars['ansible_host'] = primary_network_ip
        else:
            hostvars['ansible_host'] = primary_network_ip

    def instance_add_host_storage_devices(self, instance, hostvars):
        glusterfs_devices = []
        for disk in instance['disks']:
            device =  '/dev/disk/by-id/google-' + disk['deviceName']
            if re.match(r'docker(-?vg)?$', disk['deviceName']):
                hostvars['container_runtime_docker_storage_setup_device'] = device
            elif disk['deviceName'].startswith('glusterfs-'):
                glusterfs_devices.append(device)
        if len(glusterfs_devices) > 0:
            hostvars['glusterfs_devices'] = glusterfs_devices

    def instance_host_vars(self, instance):
        hostvars = {}
        self.instance_add_ip_vars(instance, hostvars)
        self.instance_add_ansible_vars(instance, hostvars)

        if instance.get('labels',{}).get('openshift-cluster-controller', 'false') != 'true':
            hostvars['openshift_node_group_name'] = \
                'node-config-' + self.instance_openshift_node_group_name(instance)
            hostvars['demo_node_labels'] = \
                self.instance_openshift_node_labels(instance)
            self.instance_add_host_storage_devices(instance, hostvars)

        return hostvars

    def ansible_group_filter(self, instance):
        if 'ANSIBLE_GROUP_FILTER' not in os.environ:
            return True
        for item in instance['metadata']['items']:
            if( item['key'] == 'ansible-host-group-' + os.environ['ANSIBLE_GROUP_FILTER']
            and item['value'] == 'true' ):
                return True
        return False

    def openshift_role_filter(self, hostvars):
        if 'OPENSHIFT_ROLE_FILTER' not in os.environ:
            return True
        for role in os.environ['OPENSHIFT_ROLE_FILTER'].split(','):
            kuberole = 'node-role.kubernetes.io/' + role
            if kuberole in hostvars.get('demo_node_labels', {}):
                return True
        return False

    def get_host(self, hostname):
        instance = self.get_instance(hostname)

        if not instance or instance['status'] != 'RUNNING':
            return
        else:
            return self.instance_host_vars(instance)

    def populate_hosts_with_instances(self, hosts):
        for instance in self.get_cluster_instances():
            # Skip instances that are not running
            if instance['status'] != 'RUNNING':
                continue
            hostvars = self.instance_host_vars(instance)

            if not self.openshift_role_filter(hostvars):
                continue

            fqdn = self.instance_fqdn(instance)
            hosts['_meta']['hostvars'][fqdn] = hostvars

            for group in self.instance_ansible_host_groups(instance):
                if group in hosts:
                    hosts[group]['hosts'].append(fqdn)
                else:
                    hosts[group] = {
                        'hosts': [fqdn]
                    }

    def populate_hosts(self, hosts):
        self.populate_hosts_with_instances(hosts)

    def wait_for_hosts_running(self, timeout):
        start_time = time.time()
        all_ready = False
        instance_name = ''
        instance_status = ''
        while timeout > time.time() - start_time:
            try:
                for instance in self.get_cluster_instances():
                    instance_name = instance['name']
                    instance_status = instance['status']
                    if instance['status'] != 'RUNNING':
                        raise Exception(
                            "Instance %s not status RUNNING" % (instance_name)
                        )
                all_ready = True
            except:
                pass
            if all_ready:
                break
            logging.info("Waiting for all instances to be RUNNING")
            time.sleep(2)
        if not all_ready:
            raise Exception("Instance %s found with status %s" % (instance_name, instance_status))

    def stop_instance(self, instance):
        instance_zone = instance['zone'].rsplit('/', 1)[1]

        self.computeAPI.instances().stop(
            instance = instance['name'],
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            zone = instance_zone
        ).execute()

        while instance['status'] != 'TERMINATED':
            time.sleep(5)
            instance = self.get_instance_in_zone(instance['name'], instance_zone)

    def delete_instance(self, instance):
        instance_zone = instance['zone'].rsplit('/', 1)[1]

        self.computeAPI.instances().delete(
            instance = instance['name'],
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            zone = instance_zone
        ).execute()

        while instance:
            time.sleep(5)
            instance = self.get_instance_in_zone(instance['name'], instance_zone)

    def create_node_image(self):
        image_instance_name = self.ocpinv().cluster_var('openshift_gcp_prefix') + 'image'
        image_name = self.ocpinv().cluster_var('openshift_gcp_prefix') + 'node-' + datetime.now().strftime('%Y%m%d%H%M%S')
        instance = self.get_instance(image_instance_name)
        if not instance:
            raise Exception("Unable to find cluster image instance.")
        self.stop_instance(instance)
        self.computeAPI.images().insert(
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            body = {
                "name": image_name,
                "family": self.ocpinv().cluster_var('demo_gcp_node_image_family'),
                "labels": {
                    "openshift-cluster": self.ocpinv().cluster_name
                },
                "sourceDisk": instance['disks'][0]['source']
            }
        ).execute()
        while True:
            image = self.computeAPI.images().get(
                image = image_name,
                project = self.ocpinv().cluster_var('openshift_gcp_project'),
            ).execute()
            if image['status'] == 'READY':
                break
        self.delete_instance(instance)

    def cleanup(self):
        self.cleanup_global_addresses()
        self.cleanup_images()

    def cleanup_global_addresses(self):
        cluster_prefix = self.ocpinv().cluster_var('openshift_gcp_prefix')
        req = self.computeAPI.globalAddresses().list(
            project = self.ocpinv().cluster_var('openshift_gcp_project')
        )
        while req:
            resp = req.execute()
            for address in resp.get('items', []):
                if address.get('name', '').startswith(cluster_prefix):
                    print("Releasing address {}".format(address['name']))
                    self.computeAPI.globalAddresses().delete(
                        project = self.ocpinv().cluster_var('openshift_gcp_project'),
                        address = address['id']
                    ).execute()
            req = self.computeAPI.instances().list_next(
                previous_request = req,
                previous_response = resp
            )

    def cleanup_images(self):
        cluster_prefix = self.ocpinv().cluster_var('openshift_gcp_prefix')
        req = self.computeAPI.images().list(
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            filter = '(family="{}")'.format(
                self.ocpinv().cluster_var('demo_gcp_node_image_family')
            )
        )
        while req:
            resp = req.execute()
            for image in resp.get('items', []):
                print("Deleting image {}".format(image['name']))
                self.computeAPI.images().delete(
                    project = self.ocpinv().cluster_var('openshift_gcp_project'),
                    image = image['id']
                ).execute()
            req = self.computeAPI.instances().list_next(
                previous_request = req,
                previous_response = resp
            )

    def init(self):
        self.init_gcp_services()

    def init_gcp_services(self):
        print("Enabling GCP APIs")
        project_number = self.get_project_number()

        service_ids = [
            'cloudresourcemanager.googleapis.com',
            'compute.googleapis.com',
            'iam.googleapis.com',
            'serviceusage.googleapis.com'
        ]
        if self.ocpinv().cluster_var('demo_use_cloud_dns'):
            service_ids.append('dns.googleapis.com')

        self.serviceusageAPI \
            .services() \
            .batchEnable(
                parent = 'projects/{}'.format(project_number),
                body = {
                    "serviceIds": service_ids
                }
            ).execute()
        print("Waiting for API availability...")
        for i in range(30):
            try:
                self.get_gcp_project()
                print("Ready!")
                return
            except googleapiclient.errors.HttpError:
                time.sleep(10)

    def get_gcp_project(self):
        return self.cloudresourcemanagerAPI \
            .projects() \
            .get(projectId=self.ocpinv().cluster_var('openshift_gcp_project')) \
            .execute()

    def get_project_number(self):
        try:
            project = self.get_gcp_project()
            return project['projectNumber']
        except googleapiclient.errors.HttpError as e:
            err_content = json.loads(e.content)
            m = re.search(
                r'project=(\d+)',
                err_content.get('error',{}).get('message','')
            )
            if m:
                return m.group(1)
            else:
                raise e

    def scaleup(self):
        node_groups = self.ocpinv().cluster_config.get('demo_openshift_node_groups', {})
        for node_group_name, node_group in node_groups.items():
            if node_group.get('static_node_group', False):
                continue
            minimum_instance_count = node_group.get(
                'minimum_instance_count',
                node_group.get('instance_count', 0)
            )
            gcp_zones = self.get_cluster_zones()
            zone_count = len(gcp_zones)
            for i, zone in enumerate(gcp_zones):
                minimum_target_size = int(
                    (minimum_instance_count + zone_count - i - 1) / zone_count
                )

                instance_group_name = (
                    self.ocpinv().cluster_var('openshift_gcp_prefix') +
                    node_group_name + zone[-2:]
                )

                self.scaleup_managed_instance_group(
                    instance_group_name,
                    zone,
                    minimum_target_size
                )

    def scaleup_managed_instance_group(
        self,
        instance_group_name,
        zone,
        minimum_target_size
    ):
        if minimum_target_size < 1:
            return

        instance_group_manager = self.computeAPI.instanceGroupManagers().get(
            project = self.ocpinv().cluster_var('openshift_gcp_project'),
            zone = zone,
            instanceGroupManager = instance_group_name
        ).execute()

        if instance_group_manager['targetSize'] < minimum_target_size:
            logging.info("Scaling up %s to %d" % (instance_group_name, minimum_target_size))
            instance_group_manager = self.computeAPI.instanceGroupManagers().resize(
                project = self.ocpinv().cluster_var('openshift_gcp_project'),
                zone = zone,
                instanceGroupManager = instance_group_name,
                size = minimum_target_size
            ).execute()
            return True