model.js 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. const uuid = require('uuid/v4');
  2. const Client = require('kubernetes-client').Client;
  3. const config = require('kubernetes-client').config;
  4. var client;
  5. try {
  6. client = new Client({ config: config.fromKubeconfig(), version: '1.10' });
  7. } catch(error) {
  8. client = new Client({ config: config.getInCluster(), version: '1.10' });
  9. }
  10. const Command = require('./command.js');
  11. const Prometheus = require('prom-client');
  12. const request = require("request");
  13. var configuration = require('./config');
  14. const TENANT = "12a0079b-1591-4ca0-b721-a446bda74e67"
  15. // Store in memory for now.
  16. const clusters = {};
  17. // Basic CRUD operations, no validation, no pagination, etc.
  18. const model = {};
  19. //impala_queries_open_running
  20. //impala_queries_open_queued
  21. //impala_queries_closed_failed
  22. //...
  23. const impalaQueriesMetrics = new Prometheus.Gauge({
  24. name: 'impala_queries',
  25. help: 'Query metrics',
  26. labelNames: ['datawarehouse', 'status']
  27. });
  28. const impalaQueriesCounter = new Prometheus.Gauge({
  29. name: 'impala_queries_count',
  30. help: 'Query metrics',
  31. labelNames: ['datawarehouse']
  32. });
  33. function createCluster(options) {
  34. const cluster = Object.assign({}, {
  35. cdhVersion: "CDH6.3",
  36. hdfsNamenodeHost: "hdfs-namenode",
  37. hdfsNamenodePort: 9820, // 8020
  38. // metastore.uris=thrift://spark2-envelope515-1.gce.cloudera.com:9083,hdfs.namenode.host=spark2-envelope515-1.gce.cloudera.com,hdfs.namenode.port=8020
  39. workerCpuCores: 2,
  40. workerMemoryInGib: 4,
  41. workerReplicas: 1,
  42. workerAutoResize: false,
  43. workercurrentCPUUtilizationPercentage: 0
  44. }, options);
  45. cluster.name = cluster.clusterName;
  46. cluster.crn = `crn:altus:dataware:k8s:${TENANT}:cluster:${cluster.clusterName}/${uuid()}`;
  47. cluster.creationDate = new Date().toISOString();
  48. cluster.status = "STARTING";
  49. cluster.workerReplicasOnline = 0;
  50. endpointHost = "impala-coordinator-" + cluster.clusterName;
  51. cluster.coordinatorEndpoint = {privateHost: endpointHost, publicHost: endpointHost, port: 21050};
  52. return cluster;
  53. }
  54. model.createCluster = async function(options) {
  55. console.log("Create cluster: " + JSON.stringify(options));
  56. const cluster = createCluster(options);
  57. clusters[cluster.crn] = cluster;
  58. await Command.runCommand(`helm install impala-engine --set-string registry=${configuration.registry},tag=${configuration.registryImpalaTag},name=${cluster.clusterName},worker.replicas=${cluster.workerReplicas},hdfs.namenode.host=${cluster.hdfsNamenodeHost},hdfs.namenode.port=${cluster.hdfsNamenodePort} -n ${cluster.clusterName} --repo=${configuration.helmRepo}`);
  59. return {"cluster": cluster};
  60. };
  61. async function updateClusterStatus(cluster) {
  62. if (cluster.status == 'TERMINATED') {
  63. return
  64. }
  65. var statefulset;
  66. try {
  67. // TODO: use labels
  68. statefulset = await client.apis.apps.v1.namespaces('default').deployments("impala-worker-" + cluster.clusterName).get()
  69. }
  70. catch(error) {
  71. console.log(error);
  72. // If killed manually?
  73. //if (clusters.status == 'TERMINATING') {
  74. cluster.status = 'TERMINATED';
  75. cluster.workerReplicasOnline = 0;
  76. return
  77. //}
  78. }
  79. if (statefulset == null) {
  80. cluster.status = "STARTING";
  81. cluster.workerReplicasOnline = 0;
  82. return
  83. }
  84. status = statefulset.body.status;
  85. if (status.readyReplicas == null) {
  86. // Statefulset when just launched will have undefined ready replicas.
  87. return
  88. }
  89. if ((cluster.status == "SCALING_UP" || cluster.status == "SCALING_DOWN") && status.replicas != cluster.workerReplicas) {
  90. // We are still in progress updating the cluster, don't update the status until statefulset replicas are updated.
  91. return
  92. }
  93. if (status.replicas == status.readyReplicas) {
  94. cluster.status = "ONLINE";
  95. }
  96. if (cluster.workerAutoResize) {
  97. const hpa = await client.apis.autoscaling.v1.namespaces('default').horizontalpodautoscalers("impala-worker-" + cluster.clusterName).get()
  98. hpaStatus = hpa.body.status;
  99. // TODO: polish a bit status when hpa not fully running yet
  100. cluster.workerReplicasOnline = hpaStatus.currentReplicas;
  101. cluster.workerReplicas = hpaStatus.currentCPUUtilizationPercentage >= 0 ? hpaStatus.desiredReplicas : hpaStatus.currentReplicas;
  102. cluster.workercurrentCPUUtilizationPercentage = hpaStatus.currentCPUUtilizationPercentage >= 0 ? hpaStatus.currentCPUUtilizationPercentage : "N/A";
  103. } else {
  104. cluster.workerReplicasOnline = status.readyReplicas;
  105. }
  106. }
  107. model.describeCluster = async function(options) {
  108. console.log("Describe cluster: " + JSON.stringify(options));
  109. const cluster = clusters[options.clusterName];
  110. if (cluster != undefined) {
  111. updateClusterStatus(cluster);
  112. return {"cluster": cluster};
  113. } // TODO else
  114. return {}
  115. }
  116. model.listClusters = async function(options) {
  117. var promises = [];
  118. var clusterNames = [];
  119. for (var crn in clusters) {
  120. promises.push(updateClusterStatus(clusters[crn]));
  121. clusterNames.push(clusters[crn].name);
  122. }
  123. var statefulsets = await client.apis.apps.v1.namespaces('default').statefulsets().get(); // Also append DW clusters created outside of provisioner
  124. statefulsets.body.items.forEach(function(statefulset) {
  125. var name = statefulset['metadata']['name'].substring("impala-coordinator-".length);
  126. if (statefulset['metadata']['name'].startsWith("impala-coordinator-") && clusterNames.indexOf(name) == -1) {
  127. var cluster = createCluster({clusterName: name});
  128. cluster.status = 'ONLINE';
  129. cluster.crn = cluster.name // For now crn is the cluster name
  130. clusters[cluster.crn] = cluster;
  131. }
  132. });
  133. Promise.all(promises);
  134. return Object.values(clusters);
  135. }
  136. model.updateCluster = async function(options) {
  137. console.log("Update cluster: " + JSON.stringify(options));
  138. var cluster = clusters[options.clusterName];
  139. var command = "";
  140. if (options.updateClusterAutoResizeChanged) {
  141. cluster.workerAutoResize = options.updateClusterAutoResize;
  142. if (options.updateClusterAutoResize) {
  143. cluster.workerAutoResizeMin = options.updateClusterAutoResizeMin;
  144. cluster.workerAutoResizeMax = options.updateClusterAutoResizeMax;
  145. cluster.workerAutoResizeCpu = options.updateClusterAutoResizeCpu;
  146. command = `kubectl autoscale deployment impala-worker-${cluster.clusterName} --min=${options.updateClusterAutoResizeMin} --max=${options.updateClusterAutoResizeMax} --cpu-percent=${options.updateClusterAutoResizeCpu}`;
  147. } else {
  148. command = `kubectl delete hpa impala-worker-${cluster.clusterName}`;
  149. }
  150. // TODO: check SCALING UP/DOWN status too
  151. } else {
  152. if (cluster.workerReplicas != options.workerReplicas) {
  153. let originalReplicas = cluster.workerReplicas;
  154. cluster.workerReplicas = options.workerReplicas;
  155. if (cluster.workerReplicas > originalReplicas) {
  156. cluster.status = "SCALING_UP";
  157. } else if (cluster.workerReplicas < originalReplicas) {
  158. cluster.status = "SCALING_DOWN";
  159. }
  160. await Command.runCommand(`helm upgrade ${cluster.clusterName} impala-engine --set-string registry=${configuration.registry},tag=${configuration.registryImpalaTag},name=${cluster.clusterName},worker.replicas=${cluster.workerReplicas},hdfs.namenode.host=${cluster.hdfsNamenodeHost},hdfs.namenode.port=${cluster.hdfsNamenodePort} --repo=${configuration.helmRepo}`);
  161. }
  162. }
  163. if (command) {
  164. await Command.runCommand(command);
  165. }
  166. return {"cluster": cluster};
  167. }
  168. model.deleteCluster = async function(options) {
  169. console.log("Delete cluster: " + JSON.stringify(options));
  170. const cluster = clusters[options.clusterName];
  171. cluster.status = "TERMINATING";
  172. await Command.runCommand(`helm delete --purge ${cluster.clusterName}`);
  173. if (options.workerAutoResize) {
  174. await Command.runCommand(`kubectl delete hpa impala-worker-${cluster.clusterName}`);
  175. }
  176. return;
  177. }
  178. model.getClusterMetrics = async function(options) {
  179. console.log("Metric cluster: " + JSON.stringify(options));
  180. var metrics = null;
  181. Object.keys(clusters).forEach(function(key) {
  182. var cluster = clusters[key];
  183. console.log("Scrapping: " + key);
  184. if (cluster.status != "TERMINATED") {
  185. request({
  186. url: `http://${cluster.coordinatorEndpoint.privateHost}:25000/metrics?json=true`,
  187. json: true
  188. }, function (error, response, body) {
  189. console.log("Scapped: " + error + " " + response);
  190. if (!error && response.statusCode === 200) {
  191. metrics = body.metric_group.child_groups.filter(group => group.name == "impala-server")[0].metrics;
  192. var metric = metrics.filter(metric => metric.name == "impala-server.num-queries-registered")[0];
  193. impalaQueriesMetrics.labels(cluster.clusterName, 'num-queries-registered').set(metric.value);
  194. var metric = metrics.filter(metric => metric.name == "impala-server.num-queries")[0];
  195. impalaQueriesCounter.labels(cluster.clusterName).set(metric.value);
  196. }
  197. });
  198. }
  199. });
  200. return {"metrics": metrics};
  201. }
  202. module.exports = model;