Revert KubernetesService. Introduce FetchKubernetesTokenService.

app/models/ci/cluster.rb

@@ -33,13 +33,15 @@ module Ci
     }
 
     def error!(reason)
-      update!(status: statuses[:errored],
-              status_reason: reason,
-              gcp_token: nil)
+      update!(status: statuses[:errored], status_reason: reason, gcp_token: nil)
     end
 
     def on_creation?
       scheduled? || creating?
     end
+
+    def api_url
+      'https://' + endpoint
+    end
   end
 end

app/models/project_services/kubernetes_service.rb

@@ -15,7 +15,6 @@ class KubernetesService < DeploymentService
   # Bearer authentication
   # TODO:  user/password auth, client certificates
   prop_accessor :token
-  attr_accessor :username, :password
 
   # Provide a custom CA bundle for self-signed deployments
   prop_accessor :ca_pem
@@ -139,15 +138,6 @@ class KubernetesService < DeploymentService
 
   TEMPLATE_PLACEHOLDER = 'Kubernetes namespace'.freeze
 
-  def read_secrets
-    kubeclient = build_kubeclient!
-
-    kubeclient.get_secrets.as_json
-  rescue KubeException => err
-    raise err unless err.error_code == 404
-    []
-  end
-
   private
 
   def kubeconfig
@@ -167,7 +157,7 @@ class KubernetesService < DeploymentService
   end
 
   def build_kubeclient!(api_path: 'api', api_version: 'v1')
-    raise "Incomplete settings" unless api_url && (token || (username && password))
+    raise "Incomplete settings" unless api_url && actual_namespace && token
 
     ::Kubeclient::Client.new(
       join_api_url(api_path),
@@ -200,11 +190,7 @@ class KubernetesService < DeploymentService
   end
 
   def kubeclient_auth_options
-    if token
-      { bearer_token: token }
-    else
-      { username: username, password: password }
-    end
+    { bearer_token: token }
   end
 
   def join_api_url(api_path)

app/services/ci/create_cluster_service.rb

 module Ci
   class CreateClusterService < BaseService
     def execute(access_token)
+      if params['machine_type'].blank?
+        params['machine_type'] = GoogleApi::CloudPlatform::Client::DEFAULT_MACHINE_TYPE
+      end
+
       project.clusters.create(
         params.merge(user: current_user,
                      status: Ci::Cluster.statuses[:scheduled],

app/services/ci/fetch_kubernetes_token_service.rb

+##
+# TODO:
+# Almost components in this class were copied from app/models/project_services/kubernetes_service.rb
+# We should dry up those classes not to repeat the same code.
+# Maybe we should have a special facility (e.g. lib/kubernetes_api) to maintain all Kubernetes API caller.
+module Ci
+  class FetchKubernetesTokenService
+    attr_reader :api_url, :ca_pem, :username, :password
+
+    def initialize(api_url, ca_pem, username, password)
+      @api_url = api_url
+      @ca_pem = ca_pem
+      @username = username
+      @password = password
+    end
+
+    def execute
+      read_secrets.each do |secret|
+        name = secret.dig('metadata', 'name')
+        if /default-token/ =~ name
+          token_base64 = secret.dig('data', 'token')
+          return Base64.decode64(token_base64) if token_base64
+        end
+      end
+    end
+
+    private
+
+    def read_secrets
+      kubeclient = build_kubeclient!
+
+      kubeclient.get_secrets.as_json
+    rescue KubeException => err
+      raise err unless err.error_code == 404
+      []
+    end
+
+    def build_kubeclient!(api_path: 'api', api_version: 'v1')
+      raise "Incomplete settings" unless api_url && username && password
+
+      ::Kubeclient::Client.new(
+        join_api_url(api_path),
+        api_version,
+        auth_options: { username: username, password: password },
+        ssl_options: kubeclient_ssl_options,
+        http_proxy_uri: ENV['http_proxy']
+      )
+    end
+
+    def join_api_url(api_path)
+      url = URI.parse(api_url)
+      prefix = url.path.sub(%r{/+\z}, '')
+
+      url.path = [prefix, api_path].join("/")
+
+      url.to_s
+    end
+
+    def kubeclient_ssl_options
+      opts = { verify_ssl: OpenSSL::SSL::VERIFY_PEER }
+
+      if ca_pem.present?
+        opts[:cert_store] = OpenSSL::X509::Store.new
+        opts[:cert_store].add_cert(OpenSSL::X509::Certificate.new(ca_pem))
+      end
+
+      opts
+    end
+  end
+end

app/services/ci/integrate_cluster_service.rb

 module Ci
   class IntegrateClusterService
     def execute(cluster, endpoint, ca_cert, token, username, password)
-      kubernetes_service ||= cluster.project.find_or_initialize_service('kubernetes')
-
       Ci::Cluster.transaction do
-        # Update service
-        kubernetes_service.attributes = {
-          active: true,
-          api_url: endpoint,
-          ca_pem: ca_cert,
-          namespace: cluster.project_namespace,
-          token: token
-        }
-
-        kubernetes_service.save!
+        kubernetes_service ||=
+          cluster.project.find_or_initialize_service('kubernetes')
 
-        # Save info in cluster record
         cluster.update!(
           enabled: true,
           service: kubernetes_service,
@@ -25,10 +14,16 @@ module Ci
           ca_cert: ca_cert,
           endpoint: endpoint,
           gcp_token: nil,
-          status: Ci::Cluster.statuses[:created]
-        )
-      end
+          gcp_operation_id: nil,
+          status: Ci::Cluster.statuses[:created])
 
+        kubernetes_service.update!(
+          active: true,
+          api_url: cluster.api_url,
+          ca_pem: ca_cert,
+          namespace: cluster.project_namespace,
+          token: token)
+      end
     rescue ActiveRecord::RecordInvalid => e
       cluster.error!("Failed to integrate cluster into kubernetes_service: #{e.message}")
     end

app/services/ci/update_cluster_service.rb

@@ -2,22 +2,18 @@ module Ci
   class UpdateClusterService < BaseService
     def execute(cluster)
       Ci::Cluster.transaction do
-        if params['enabled'] == 'true'
+        cluster.update!(enabled: params['enabled'])
 
-          cluster.service.attributes = {
+        if params['enabled'] == 'true'
+          cluster.service.update!(
             active: true,
-            api_url: cluster.endpoint,
+            api_url: cluster.api_url,
             ca_pem: cluster.ca_cert,
             namespace: cluster.project_namespace,
-            token: cluster.kubernetes_token
-          }
-
-          cluster.service.save!
+            token: cluster.kubernetes_token)
         else
           cluster.service.update(active: false)
         end
-
-        cluster.update!(enabled: params['enabled'])
       end
     rescue ActiveRecord::RecordInvalid => e
       cluster.errors.add(:base, e.message)

app/views/projects/clusters/_form.html.haml

@@ -33,4 +33,4 @@
           = field.submit s_('ClusterIntegration|Create cluster'), class: 'btn btn-save'
 
           -# TODO: Remove before merge
-          = link_to "Create on Google Container Engine", namespace_project_clusters_path(@project.namespace, @project, cluster: {cluster_name: "gke-test-creation#{Random.rand(100)}", gcp_project_id: 'gitlab-internal-153318', cluster_zone: 'us-central1-a', cluster_size: '1', project_namespace: 'aaa', machine_type: '???'}), method: :post
\ No newline at end of file
+          = link_to "Create on Google Container Engine", namespace_project_clusters_path(@project.namespace, @project, cluster: {cluster_name: "gke-test-creation#{Random.rand(100)}", gcp_project_id: 'gitlab-internal-153318', cluster_zone: 'us-central1-a', cluster_size: '1', project_namespace: 'aaa', machine_type: 'n1-standard-1'}), method: :post
\ No newline at end of file

app/workers/wait_for_cluster_creation_worker.rb

@@ -17,69 +17,56 @@ class WaitForClusterCreationWorker
       GoogleApi::CloudPlatform::Client.new(cluster.gcp_token, nil)
 
     operation = api_client.projects_zones_operations(
-        cluster.gcp_project_id,
-        cluster.cluster_zone,
-        cluster.gcp_operation_id
-      )
+      cluster.gcp_project_id,
+      cluster.cluster_zone,
+      cluster.gcp_operation_id)
 
     if operation.is_a?(StandardError)
       return cluster.error!("Failed to request to CloudPlatform; #{operation.message}")
     end
 
     case operation.status
-    when 'DONE'
-      integrate(api_client, cluster)
     when 'RUNNING'
       if Time.now < operation.start_time.to_time + TIMEOUT
         WaitForClusterCreationWorker.perform_in(EAGER_INTERVAL, cluster.id)
       else
         return cluster.error!("Cluster creation time exceeds timeout; #{TIMEOUT}")
       end
+    when 'DONE'
+      integrate(cluster, api_client)
     else
-      return cluster.error!("Unexpected operation status; #{operation.status_message}")
+      return cluster.error!("Unexpected operation status; #{operation.status} #{operation.status_message}")
     end
   end
 
-  def integrate(api_client, cluster)
-    # Get cluster details (end point, etc)
+  def integrate(cluster, api_client)
     gke_cluster = api_client.projects_zones_clusters_get(
-        cluster.gcp_project_id,
-        cluster.cluster_zone,
-        cluster.cluster_name
-      )
+      cluster.gcp_project_id,
+      cluster.cluster_zone,
+      cluster.cluster_name)
 
     if gke_cluster.is_a?(StandardError)
       return cluster.error!("Failed to request to CloudPlatform; #{gke_cluster.message}")
     end
 
-    # Get k8s token
-    kubernetes_token = ''
-    KubernetesService.new.tap do |ks|
-      ks.api_url = 'https://' + gke_cluster.endpoint
-      ks.ca_pem = Base64.decode64(gke_cluster.master_auth.cluster_ca_certificate)
-      ks.username = gke_cluster.master_auth.username
-      ks.password = gke_cluster.master_auth.password
-      secrets = ks.read_secrets
-      secrets.each do |secret|
-        name = secret.dig('metadata', 'name')
-        if /default-token/ =~ name
-          token_base64 = secret.dig('data', 'token')
-          kubernetes_token = Base64.decode64(token_base64)
-          break
-        end
-      end
+    begin
+      endpoint = gke_cluster.endpoint
+      api_url = 'https://' + endpoint
+      ca_cert = Base64.decode64(gke_cluster.master_auth.cluster_ca_certificate)
+      username = gke_cluster.master_auth.username
+      password = gke_cluster.master_auth.password
+    rescue Exception => e
+      return cluster.error!("Can not extract the extected data; #{e}")
     end
 
+    kubernetes_token = Ci::FetchKubernetesTokenService.new(
+      api_url, ca_cert, username, password).execute
+
     unless kubernetes_token
       return cluster.error!('Failed to get a default token of kubernetes')
     end
 
-    # k8s endpoint, ca_cert
-    endpoint = 'https://' + gke_cluster.endpoint
-    ca_cert = Base64.decode64(gke_cluster.master_auth.cluster_ca_certificate)
-    username = gke_cluster.master_auth.username
-    password = gke_cluster.master_auth.password
-
-    Ci::IntegrateClusterService.new.execute(cluster, endpoint, ca_cert, kubernetes_token, username, password)
+    Ci::IntegrateClusterService.new.execute(
+      cluster, endpoint, ca_cert, kubernetes_token, username, password)
   end
 end

lib/google_api/cloud_platform/client.rb

@@ -3,6 +3,8 @@ require 'google/apis/container_v1'
 module GoogleApi
   module CloudPlatform
     class Client < GoogleApi::Auth
+      DEFAULT_MACHINE_TYPE = 'n1-standard-1'
+
       class << self
         def session_key_for_token
           :cloud_platform_access_token
@@ -27,8 +29,6 @@ module GoogleApi
         cluster
       end
 
-      # Responce exmaple 
-      # TODO: machine_type : Defailt 3.75 GB
       def projects_zones_clusters_create(project_id, zone, cluster_name, cluster_size, machine_type:)
         service = Google::Apis::ContainerV1::ContainerService.new
         service.authorization = access_token
@@ -37,7 +37,10 @@ module GoogleApi
             {
               "cluster": {
                 "name": cluster_name,
-                "initial_node_count": cluster_size
+                "initial_node_count": cluster_size,
+                "node_config": {
+                  "machine_type": machine_type # Default 3.75 GB, if ommit
+                }
               }
             }
           )