You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by ma...@apache.org on 2017/07/31 20:15:28 UTC
[incubator-openwhisk] branch master updated: Remove consul and
registrator from the deployment. (#2452)
This is an automated email from the ASF dual-hosted git repository.
markusthoemmes pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git
The following commit(s) were added to refs/heads/master by this push:
new e466e48 Remove consul and registrator from the deployment. (#2452)
e466e48 is described below
commit e466e48a8cb7bf44b358e38728f58360b8377f7e
Author: rodric rabbah <ro...@gmail.com>
AuthorDate: Mon Jul 31 16:15:26 2017 -0400
Remove consul and registrator from the deployment. (#2452)
- Update controller config.
- Update params for invoker and clean up some deadcode.
- Hardcoded db prefix for docker-machine since db init runs on host not inside VM.
---
ansible/README.md | 2 +-
ansible/consul.yml | 10 -
ansible/environments/distributed/group_vars/all | 4 -
ansible/environments/distributed/hosts | 2 -
ansible/environments/docker-machine/group_vars/all | 3 +-
ansible/environments/docker-machine/hosts.j2.ini | 4 -
ansible/environments/local/hosts | 4 -
ansible/environments/mac/hosts | 4 -
ansible/group_vars/all | 11 -
ansible/openwhisk.yml | 2 -
ansible/roles/consul/handlers/main.yml | 7 -
ansible/roles/consul/tasks/clean.yml | 34 ---
ansible/roles/consul/tasks/deploy.yml | 92 ------
ansible/roles/consul/tasks/main.yml | 10 -
ansible/roles/consul/templates/config.json.j2 | 15 -
ansible/roles/controller/tasks/deploy.yml | 37 ++-
ansible/roles/invoker/tasks/deploy.yml | 36 ++-
ansible/tasks/writeWhiskProperties.yml | 2 +-
ansible/templates/whisk.properties.j2 | 4 -
.../src/main/scala/whisk/common/ConsulClient.scala | 325 ---------------------
.../main/scala/whisk/common/ConsulKVReporter.scala | 68 -----
.../src/main/scala/whisk/core/WhiskConfig.scala | 43 +--
.../main/scala/whisk/core/controller/Backend.scala | 3 -
.../scala/whisk/core/controller/Controller.scala | 1 -
.../scala/whisk/core/controller/RestAPIs.scala | 7 -
.../core/entitlement/ActivationThrottler.scala | 6 +-
.../scala/whisk/core/entitlement/Entitlement.scala | 4 +-
.../core/loadBalancer/InvokerSupervision.scala | 18 +-
.../core/loadBalancer/LoadBalancerService.scala | 8 +-
.../scala/whisk/core/container/ContainerPool.scala | 4 -
.../whisk/core/container/ContainerUtils.scala | 3 +-
.../containerpool/docker/DockerContainer.scala | 2 +-
.../main/scala/whisk/core/invoker/Invoker.scala | 13 +-
docs/about.md | 12 +-
docs/reference.md | 1 -
tests/src/test/scala/common/WhiskProperties.java | 8 -
.../scala/whisk/consul/ConsulClientTests.scala | 198 -------------
.../scala/whisk/consul/ConsulHealthTests.scala | 61 ----
.../test/scala/whisk/core/WhiskConfigTests.scala | 23 --
.../docker/test/DockerContainerTests.scala | 2 +-
.../controller/test/ControllerTestCommon.scala | 2 -
.../test/InvokerSupervisionTests.scala | 42 +--
tools/build/redo | 4 -
tools/health/isAlive | 14 -
tools/health/kvstore | 144 ---------
45 files changed, 91 insertions(+), 1208 deletions(-)
diff --git a/ansible/README.md b/ansible/README.md
index 2d30324..176665f 100644
--- a/ansible/README.md
+++ b/ansible/README.md
@@ -204,7 +204,7 @@ cd ansible
ansible-playbook -i environments/<environment> controller.yml -e mode=clean
```
-**Caveat:** In distributed environments some components (e.g. Consul, Invoker, etc.) exist on multiple machines. So if you run a playbook to clean or deploy those components, it will run on **all** of the hosts targeted by the component's playbook.
+**Caveat:** In distributed environments some components (e.g. Invoker, etc.) exist on multiple machines. So if you run a playbook to clean or deploy those components, it will run on **all** of the hosts targeted by the component's playbook.
### Cleaning an OpenWhisk Deployment
diff --git a/ansible/consul.yml b/ansible/consul.yml
deleted file mode 100644
index 0e81cd2..0000000
--- a/ansible/consul.yml
+++ /dev/null
@@ -1,10 +0,0 @@
----
-# This playbook deploys a Consul cluster for Openwhisk.
-# Consul agents are installed on all hosts except the ansible host.
-# Consul servers are installed on all hosts in the consul_servers group.
-
-- include: properties.yml
-
-- hosts: all:!ansible
- roles:
- - consul
diff --git a/ansible/environments/distributed/group_vars/all b/ansible/environments/distributed/group_vars/all
index dfea9dc..7d30485 100755
--- a/ansible/environments/distributed/group_vars/all
+++ b/ansible/environments/distributed/group_vars/all
@@ -56,10 +56,6 @@ instances:
num_instances: 1
flavor:
- - name: consul_servers
- num_instances: 1
- flavor:
-
- name: invokers
num_instances: 2
flavor:
diff --git a/ansible/environments/distributed/hosts b/ansible/environments/distributed/hosts
index 23d4cea..2a56b7e 100755
--- a/ansible/environments/distributed/hosts
+++ b/ansible/environments/distributed/hosts
@@ -14,8 +14,6 @@ edge
10.3.2.155
[kafka]
10.3.2.156
-[consul_servers]
-10.3.2.157
[invokers]
10.3.2.158
10.3.2.159
diff --git a/ansible/environments/docker-machine/group_vars/all b/ansible/environments/docker-machine/group_vars/all
index d9b8cc6..82a1416 100644
--- a/ansible/environments/docker-machine/group_vars/all
+++ b/ansible/environments/docker-machine/group_vars/all
@@ -9,7 +9,8 @@ docker_dns: ""
# a hostname that is resolved on the client, via /etc/hosts for example.
whisk_api_localhost_name: "openwhisk"
-db_prefix: "{{ ansible_user_id|lower }}_{{ ansible_hostname|lower }}_"
+# Hardcoded for docker-machine since db init runs on host not inside VM
+db_prefix: dockermachine_
# Auto lookup to find the db credentials
db_provider: "{{ lookup('ini', 'db_provider section=db_creds file={{ playbook_dir }}/db_local.ini') }}"
diff --git a/ansible/environments/docker-machine/hosts.j2.ini b/ansible/environments/docker-machine/hosts.j2.ini
index 75d7f64..1d78b33 100644
--- a/ansible/environments/docker-machine/hosts.j2.ini
+++ b/ansible/environments/docker-machine/hosts.j2.ini
@@ -14,10 +14,6 @@ ansible ansible_connection=local
[kafka]
{{ docker_machine_ip }}
-; the consul_servers group has maximum 5 machines
-[consul_servers]
-{{ docker_machine_ip }}
-
[invokers]
{{ docker_machine_ip }}
diff --git a/ansible/environments/local/hosts b/ansible/environments/local/hosts
index 0cbc0d5..358310e 100644
--- a/ansible/environments/local/hosts
+++ b/ansible/environments/local/hosts
@@ -14,10 +14,6 @@ ansible ansible_connection=local
[kafka]
172.17.0.1 ansible_connection=local
-; the consul_servers group has maximum 5 machines
-[consul_servers]
-172.17.0.1 ansible_connection=local
-
[invokers]
172.17.0.1 ansible_connection=local
diff --git a/ansible/environments/mac/hosts b/ansible/environments/mac/hosts
index 0cbc0d5..358310e 100644
--- a/ansible/environments/mac/hosts
+++ b/ansible/environments/mac/hosts
@@ -14,10 +14,6 @@ ansible ansible_connection=local
[kafka]
172.17.0.1 ansible_connection=local
-; the consul_servers group has maximum 5 machines
-[consul_servers]
-172.17.0.1 ansible_connection=local
-
[invokers]
172.17.0.1 ansible_connection=local
diff --git a/ansible/group_vars/all b/ansible/group_vars/all
index 4a8ff4e..3c6a81b 100644
--- a/ansible/group_vars/all
+++ b/ansible/group_vars/all
@@ -112,17 +112,6 @@ controller:
blackboxFraction: 0.10
instances: "{{ groups['controllers'] | length }}"
-consul:
- confdir: "{{ config_root_dir }}/consul"
- version: 0.7.0
- port:
- server: 8300
- serf_lan: 8301
- serf_wan: 8302
- rpc: 8400
- http: 8500
- dns: 8600
-
registry:
confdir: "{{ config_root_dir }}/registry"
diff --git a/ansible/openwhisk.yml b/ansible/openwhisk.yml
index 94dfcfe..58263b6 100644
--- a/ansible/openwhisk.yml
+++ b/ansible/openwhisk.yml
@@ -3,8 +3,6 @@
# It assumes you have already set up your database with the respective db provider playbook (currently cloudant.yml or couchdb.yml)
# It assumes that wipe.yml have being deployed at least once
-- include: consul.yml
-
- include: kafka.yml
- include: controller.yml
diff --git a/ansible/roles/consul/handlers/main.yml b/ansible/roles/consul/handlers/main.yml
deleted file mode 100644
index 0fdb3e0..0000000
--- a/ansible/roles/consul/handlers/main.yml
+++ /dev/null
@@ -1,7 +0,0 @@
----
-# This handler contains common tasks (aka. utilities) for deploying OpenWhisk
-# If a handler is notified by multiple tasks, it will still be run only once
-
-# TODO: avoid py script? persist a file in the role would be better
-- name: fill consul kv
- local_action: command python "{{ openwhisk_home }}/tools/health/kvstore" --import -d "{{ playbook_dir }}/../"
\ No newline at end of file
diff --git a/ansible/roles/consul/tasks/clean.yml b/ansible/roles/consul/tasks/clean.yml
deleted file mode 100644
index d8dc72a..0000000
--- a/ansible/roles/consul/tasks/clean.yml
+++ /dev/null
@@ -1,34 +0,0 @@
----
-# Remove consul agent/server and registrator containers as well as the consul.confdir
-
-- name: remove registrator
- docker_container:
- name: registrator
- image: "gliderlabs/registrator"
- state: absent
- ignore_errors: True
-
-- name: remove consul
- docker_container:
- name: consul
- image: "{{ docker_registry }}{{ docker.image.prefix }}/consul:{{ docker.image.tag }}"
- state: absent
- ignore_errors: True
-
-- name: remove consul config directory
- file:
- path: "{{ consul.confdir }}"
- state: absent
- become: true
-
-- name: remove consul log directory
- file:
- path: "{{ whisk_logs_dir }}/consul"
- state: absent
- become: true
-
-- name: remove registrator log directory
- file:
- path: "{{ whisk_logs_dir }}/registrator"
- state: absent
- become: true
diff --git a/ansible/roles/consul/tasks/deploy.yml b/ansible/roles/consul/tasks/deploy.yml
deleted file mode 100644
index 8a1283f..0000000
--- a/ansible/roles/consul/tasks/deploy.yml
+++ /dev/null
@@ -1,92 +0,0 @@
----
-# This role will install Consul Servers/Agents in all machines. After that it installs the Registrators.
-# There is a group of machines in the corresponding environment inventory called 'consul_servers' where the Consul Servers are installed
-# In this way they build up a Consul Cluster
-# Other machines that are not in the 'consul_servers' group, have the Consul Agents
-# The template 'config.json.j2' will look at the environment inventory to decide to generate a config file for booting a server or an agent
-
-- name: ensure consul config directory exists
- file:
- path: "{{ consul.confdir }}"
- state: directory
- when: "'consul_servers' in group_names"
-
-- name: copy template from local to remote in consul config directory
- template:
- src: config.json.j2
- dest: "{{ consul.confdir }}/config.json"
- when: "'consul_servers' in group_names"
-
-- name: ensure consul log directory is created with permissions
- file:
- path: "{{ whisk_logs_dir }}/consul"
- state: directory
- mode: 0777
- become: true
-
-- name: "pull the consul:{{ consul.version }} image"
- shell: "docker pull consul:{{ consul.version }}"
- retries: "{{ docker.pull.retries }}"
- delay: "{{ docker.pull.delay }}"
-
-- name: (re)start consul server/agent
- docker_container:
- name: consul
- image: consul:{{ consul.version }}
- state: started
- recreate: true
- restart_policy: "{{ docker.restart.policy }}"
- command: "consul agent -config-dir /consul/config"
- volumes:
- - "{{ whisk_logs_dir }}/consul:/logs"
- - "{{ consul.confdir }}:/consul/config"
- ports:
- - "{{ consul.port.dns }}:8600/udp"
- - "{{ consul.port.http }}:8500"
- - "{{ consul.port.rpc }}:8400"
- - "{{ consul.port.serf_wan }}:8302"
- - "{{ consul.port.serf_lan }}:8301"
- - "{{ consul.port.serf_wan }}:8302/udp"
- - "{{ consul.port.serf_lan }}:8301/udp"
- - "{{ consul.port.server }}:8300"
- when: "'consul_servers' in group_names"
-
-- name: wait until the Consul Server/Agent in this host is up and running
- uri:
- method: PUT
- url: "http://{{ inventory_hostname }}:{{ consul.port.http }}/v1/kv/consulIsAlive"
- body: 'true'
- register: result
- until: result.status == 200
- retries: 12
- delay: 5
- when: "'consul_servers' in group_names"
-
-- name: delete is alive token from Consul Server/Agent
- uri:
- method: DELETE
- url: "http://{{ inventory_hostname }}:{{ consul.port.http }}/v1/kv/consulIsAlive"
- register: result
- until: result.status == 200
- retries: 10
- delay: 1
- when: "'consul_servers' in group_names"
-
-- name: notify handler to fill in Consul KV store with parameters in whisk.properties
- command: "true"
- notify: fill consul kv
- when: "'consul_servers' in group_names"
-
-- name: start registrator using docker cli
- shell: >
- docker run -d
- --name registrator
- --hostname registrator
- --restart {{ docker.restart.policy }}
- --log-driver syslog
- -v {{ whisk_logs_dir }}/registrator:/logs
- -v {{ docker_sock | default('/var/run/docker.sock') }}:/tmp/docker.sock
- --userns host
- gliderlabs/registrator -ip {{ ansible_host | default(inventory_hostname) }} -resync 2 consul://{{ groups['consul_servers'] | first }}:{{ consul.port.http }}
-
-# todo: re-enable docker_container module once https://github.com/ansible/ansible-modules-core/issues/5054 is resolved
diff --git a/ansible/roles/consul/tasks/main.yml b/ansible/roles/consul/tasks/main.yml
deleted file mode 100644
index 7259b27..0000000
--- a/ansible/roles/consul/tasks/main.yml
+++ /dev/null
@@ -1,10 +0,0 @@
----
-# This role will install Consul Servers/Agents in all machines.
-# In deploy mode it will deploy a consul agent or server depending on whether the host is in the consul_servers group.
-# In clean mode it will remove consul agents/server and registrator containers as well as consul.confdir
-
-- include: deploy.yml
- when: mode == "deploy"
-
-- include: clean.yml
- when: mode == "clean"
diff --git a/ansible/roles/consul/templates/config.json.j2 b/ansible/roles/consul/templates/config.json.j2
deleted file mode 100644
index b0e6131..0000000
--- a/ansible/roles/consul/templates/config.json.j2
+++ /dev/null
@@ -1,15 +0,0 @@
-{# this template is used to generate a config.json for booting a consul server #}
-{
- "server": true,
- "data_dir": "/consul/data",
- "ui": true,
- "log_level": "WARN",
- "node_name": "{{ ansible_host | default(inventory_hostname) }}",
- "client_addr": "0.0.0.0",
- "advertise_addr": "{{ ansible_host | default(inventory_hostname) }}",
- "ports": {
- "dns": 8600
- },
- "bootstrap": true,
- "disable_update_check": true
-}
\ No newline at end of file
diff --git a/ansible/roles/controller/tasks/deploy.yml b/ansible/roles/controller/tasks/deploy.yml
index a663689..3f62ca9 100644
--- a/ansible/roles/controller/tasks/deploy.yml
+++ b/ansible/roles/controller/tasks/deploy.yml
@@ -25,18 +25,41 @@
restart_policy: "{{ docker.restart.policy }}"
hostname: "controller{{ groups['controllers'].index(inventory_hostname) }}"
env:
+ "JAVA_OPTS": "-Xmx{{ controller.heap }}"
+ "CONTROLLER_OPTS": "{{ controller.arguments }}"
+
"COMPONENT_NAME": "controller{{ groups['controllers'].index(inventory_hostname) }}"
- "CONSULSERVER_HOST": "{{ groups['consul_servers'] | first }}"
- "CONSUL_HOST_PORT4": "{{ consul.port.http }}"
"PORT": 8080
+
"WHISK_VERSION_NAME": "{{ whisk_version_name }}"
"WHISK_VERSION_DATE": "{{ whisk.version.date }}"
"WHISK_VERSION_BUILDNO": "{{ docker.image.tag }}"
- "SERVICE_CHECK_HTTP": "/ping"
- "SERVICE_CHECK_TIMEOUT": "2s"
- "SERVICE_CHECK_INTERVAL": "15s"
- "JAVA_OPTS": "-Xmx{{ controller.heap }}"
- "CONTROLLER_OPTS": "{{ controller.arguments }}"
+
+ "KAFKA_HOST": "{{ groups['kafka']|first }}"
+ "KAFKA_HOST_PORT": "{{ kafka.port }}"
+
+ "DB_PROTOCOL": "{{ db_protocol }}"
+ "DB_PROVIDER": "{{ db_provider }}"
+ "DB_HOST": "{{ db_host }}"
+ "DB_PORT": "{{ db_port }}"
+ "DB_USERNAME": "{{ db_username }}"
+ "DB_PASSWORD": "{{ db_password }}"
+ "DB_WHISK_ACTIONS": "{{ db.whisk.actions }}"
+ "DB_WHISK_AUTHS": "{{ db.whisk.auth }}"
+ "DB_WHISK_ACTIVATIONS": "{{ db.whisk.activations }}"
+
+ "DEFAULTLIMITS_ACTIONS_INVOKES_PERMINUTE": "{{ defaultLimits.actions.invokes.perMinute }}"
+ "DEFAULTLIMITS_ACTIONS_INVOKES_CONCURRENT": "{{ defaultLimits.actions.invokes.concurrent }}"
+ "DEFAULTLIMITS_TRIGGERS_FIRES_PERMINUTE": "{{ defaultLimits.triggers.fires.perMinute }}"
+ "DEFAULTLIMITS_ACTIONS_INVOKES_CONCURRENTINSYSTEM": "{{ defaultLimits.actions.invokes.concurrentInSystem }}"
+ "DEFAULTLIMITS_ACTIONS_SEQUENCE_MAXLENGTH": "{{ defaultLimits.actions.sequence.maxLength }}"
+ "LIMITS_ACTIONS_INVOKES_PERMINUTE": "{{ limits.actions.invokes.perMinute }}"
+ "LIMITS_ACTIONS_INVOKES_CONCURRENT": "{{ limits.actions.invokes.concurrent }}"
+ "LIMITS_ACTIONS_INVOKES_CONCURRENTINSYSTEM": "{{ limits.actions.invokes.concurrentInSystem }}"
+ "LIMITS_TRIGGERS_FIRES_PERMINUTE": "{{ limits.triggers.fires.perMinute }}"
+ "LOADBALANCER_INVOKERBUSYTHRESHOLD": "{{ invoker.busyThreshold }}"
+
+ "RUNTIMES_MANIFEST": "{{ runtimesManifest | to_json }}"
volumes:
- "{{ whisk_logs_dir }}/controller{{ groups['controllers'].index(inventory_hostname) }}:/logs"
ports:
diff --git a/ansible/roles/invoker/tasks/deploy.yml b/ansible/roles/invoker/tasks/deploy.yml
index 76a21d5..5c67aa2 100644
--- a/ansible/roles/invoker/tasks/deploy.yml
+++ b/ansible/roles/invoker/tasks/deploy.yml
@@ -66,16 +66,38 @@
--name invoker{{ groups['invokers'].index(inventory_hostname) }}
--hostname invoker{{ groups['invokers'].index(inventory_hostname) }}
--restart {{ docker.restart.policy }}
+ -e JAVA_OPTS=-Xmx{{ invoker.heap }}
+ -e INVOKER_OPTS='{{ invoker.arguments }}'
-e COMPONENT_NAME=invoker{{ groups['invokers'].index(inventory_hostname) }}
- -e CONSULSERVER_HOST={{ groups['consul_servers'] | first }}
- -e CONSUL_HOST_PORT4={{ consul.port.http }}
-e PORT=8080
+ -e KAFKA_HOST={{ groups['kafka']|first }}
+ -e KAFKA_HOST_PORT={{ kafka.port }}
+ -e DB_PROTOCOL={{ db_protocol }}
+ -e DB_PROVIDER={{ db_provider }}
+ -e DB_HOST={{ db_host }}
+ -e DB_PORT={{ db_port }}
+ -e DB_USERNAME={{ db_username }}
+ -e DB_PASSWORD={{ db_password }}
+ -e DB_WHISK_ACTIONS={{ db.whisk.actions }}
+ -e DB_WHISK_ACTIVATIONS={{ db.whisk.activations }}
+ -e WHISK_API_HOST_PROTO={{ whisk_api_host_proto | default('https') }}
+ -e WHISK_API_HOST_PORT={{ whisk_api_host_port | default('443') }}
+ -e WHISK_API_HOST_NAME={{ whisk_api_host_name | default(groups['edge'] | first) }}
+ -e RUNTIMES_MANIFEST='{{ runtimesManifest | to_json }}'
-e SELF_DOCKER_ENDPOINT=localhost
- -e SERVICE_CHECK_HTTP=/ping
- -e SERVICE_CHECK_TIMEOUT=2s
- -e SERVICE_CHECK_INTERVAL=15s
- -e JAVA_OPTS=-Xmx{{ invoker.heap }}
- -e INVOKER_OPTS='{{ invoker.arguments }}'
+ -e DOCKER_REGISTRY={{ docker_registry }}
+ -e DOCKER_IMAGE_PREFIX={{ docker.image.prefix }}
+ -e DOCKER_IMAGE_TAG={{ docker.image.tag }}
+ -e INVOKER_CONTAINER_NETWORK={{ invoker_container_network_name | default("bridge") }}
+ -e INVOKER_CONTAINER_POLICY={{ invoker_container_policy_name | default()}}
+ -e INVOKER_CONTAINER_DNS={{ invoker_container_network_dns_servers | default()}}
+ -e INVOKER_NUMCORE={{ invoker.numcore }}
+ -e INVOKER_CORESHARE={{ invoker.coreshare }}
+ -e INVOKER_SERIALIZEDOCKEROP={{ invoker.serializeDockerOp }}
+ -e INVOKER_SERIALIZEDOCKERPULL={{ invoker.serializeDockerPull }}
+ -e INVOKER_USERUNC={{ invoker_use_runc | default(invoker.useRunc) }}
+ -e INVOKER_USEREACTIVEPOOL={{ invoker.useReactivePool }}
+ -e WHISK_LOGS_DIR={{ whisk_logs_dir }}
-v /sys/fs/cgroup:/sys/fs/cgroup
-v /run/runc:/run/runc
-v {{ whisk_logs_dir }}/invoker{{ groups['invokers'].index(inventory_hostname) }}:/logs
diff --git a/ansible/tasks/writeWhiskProperties.yml b/ansible/tasks/writeWhiskProperties.yml
index a2a8d81..4f5fc14 100644
--- a/ansible/tasks/writeWhiskProperties.yml
+++ b/ansible/tasks/writeWhiskProperties.yml
@@ -1,6 +1,6 @@
---
# This task will write whisk.properties to the openwhisk_home.
-# Currently whisk.properties is still needed for consul and tests.
+# Currently whisk.properties is still needed for tests.
- name: write whisk.properties template to openwhisk_home
template:
diff --git a/ansible/templates/whisk.properties.j2 b/ansible/templates/whisk.properties.j2
index 728bff6..af5f908 100644
--- a/ansible/templates/whisk.properties.j2
+++ b/ansible/templates/whisk.properties.j2
@@ -47,7 +47,6 @@ limits.actions.invokes.concurrentInSystem={{ limits.actions.invokes.concurrentIn
limits.triggers.fires.perMinute={{ limits.triggers.fires.perMinute }}
{% endif %}
-consulserver.host={{ groups["consul_servers"]|first }}
edge.host={{ groups["edge"]|first }}
kafka.host={{ groups["kafka"]|first }}
router.host={{ groups["edge"]|first }}
@@ -58,8 +57,6 @@ edge.host.apiport=443
zookeeper.host.port={{ zookeeper.port }}
kafka.host.port={{ kafka.port }}
kafkaras.host.port={{ kafka.ras.port }}
-consul.host.port4={{ consul.port.http }}
-consul.host.port5={{ consul.port.server }}
invoker.hosts.baseport={{ invoker.port }}
controller.hosts={{ groups["controllers"] | join(",") }}
@@ -77,7 +74,6 @@ invoker.useRunc={{ invoker_use_runc | default(invoker.useRunc) }}
invoker.useReactivePool={{ invoker.useReactivePool }}
invoker.instances={{ invoker.instances }}
-consulserver.docker.endpoint={{ groups["consul_servers"]|first }}:{{ docker.port }}
edge.docker.endpoint={{ groups["edge"]|first }}:{{ docker.port }}
kafka.docker.endpoint={{ groups["kafka"]|first }}:{{ docker.port }}
main.docker.endpoint={{ groups["controllers"]|first }}:{{ docker.port }}
diff --git a/common/scala/src/main/scala/whisk/common/ConsulClient.scala b/common/scala/src/main/scala/whisk/common/ConsulClient.scala
deleted file mode 100644
index d45667c..0000000
--- a/common/scala/src/main/scala/whisk/common/ConsulClient.scala
+++ /dev/null
@@ -1,325 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-package whisk.common
-
-import scala.concurrent.Future
-
-import org.apache.commons.codec.binary.Base64
-
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
-import akka.http.scaladsl.model._
-import akka.http.scaladsl.unmarshalling._
-import akka.stream.ActorMaterializer
-
-import java.util.NoSuchElementException
-
-import akka.stream.scaladsl._
-import spray.json._
-import spray.json.DefaultJsonProtocol._
-import scala.util.Try
-
-/**
- * Client to access Consul's <a href="https://www.consul.io/docs/agent/http.html">
- * HTTP API</a>
- */
-class ConsulClient(hostAndPort: String)(implicit val actorSystem: ActorSystem) {
- // A consequence of hostAndPort being merged in config.
- private val host :: port :: Nil = hostAndPort.split(":").toList
-
- private implicit val executionContext = actorSystem.dispatcher
- private implicit val materializer = ActorMaterializer()
-
- private val base = Uri().withScheme("http").withHost(host).withPort(port.toInt)
-
- val kv = new ConsulKeyValueApi(base)
- val health = new ConsulHealthApi(base)
- val catalog = new ConsulCatalogApi(base)
-}
-
-/*
- * Consul API for the key/value store
- */
-
-case class ConsulEntry(key: String, value: Option[String]) {
- val decodedValue = value map { value =>
- new String(Base64.decodeBase64(value))
- }
-}
-object ConsulEntry extends DefaultJsonProtocol {
- // Consul's JSON responses have capitalized keynames, scala standard
- // is lowercased fields though, so we explicitly name the parameters
- // for ConsulEntry here.
- implicit val serdes = jsonFormat(ConsulEntry.apply, "Key", "Value")
-}
-
-/**
- * Client to access Consul's Key/Value-Store API
- */
-class ConsulKeyValueApi(base: Uri)(implicit val actorSystem: ActorSystem, val materializer: ActorMaterializer) extends KeyValueStore {
- private implicit val executionContext = actorSystem.dispatcher
-
- private def uriWithKey(key: String) = base.withPath(Uri.Path(s"/v1/kv/$key"))
-
- override def get(key: String): Future[String] = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.GET,
- uri = uriWithKey(key)))
- r.flatMap { response =>
- if (response.status == StatusCodes.OK) {
- Unmarshal(response.entity).to[List[ConsulEntry]]
- } else {
- response.entity.dataBytes.runWith(Sink.ignore)
- Future.failed(new NoSuchElementException())
- }
- } map { _.head.decodedValue.getOrElse(throw new NoSuchElementException()) }
- }
-
- override def put(key: String, value: String): Future[Any] = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.PUT,
- uri = uriWithKey(key),
- entity = value))
- r.flatMap { response =>
- Unmarshal(response).to[Any]
- }
- }
-
- override def del(key: String): Future[Any] = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.DELETE,
- uri = uriWithKey(key)))
- r.flatMap { response =>
- Unmarshal(response).to[Any]
- }
- }
-
- /**
- * Gets all entries in a path
- *
- * @param root the key under which to get the entries
- * @return a future that completes with key/value pairs
- */
- def getRecurse(root: String): Future[Map[String, String]] = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.GET,
- uri = uriWithKey(root).withQuery(Uri.Query("recurse" -> "true"))))
- r.flatMap { response =>
- if (response.status != StatusCodes.NotFound) {
- Unmarshal(response).to[List[ConsulEntry]]
- } else {
- Future.failed(new NoSuchElementException())
- }
- } map { entries =>
- entries.map(e => e.key -> e.decodedValue.getOrElse("")).toMap
- }
- }
-}
-
-trait KeyValueStore {
- /**
- * Gets the entry from the key/value store with the given key
- *
- * @param key the key to get
- * @return a future that completes the value from the store
- */
- def get(key: String): Future[String]
-
- /**
- * Writes the key/value entry with the given key.
- * If the key already exists, the value is overridden
- *
- * @param key the key to store
- * @param value the value to store
- * @return a future that completes on success of the operation
- */
- def put(key: String, value: String): Future[Any]
-
- /**
- * Deletes the entry with the given key from the
- * key/value store
- *
- * @param key the key to delete
- * @return a future that completes on success of the operation
- */
- def del(key: String): Future[Any]
-}
-
-/*
- * Consul Services API
- */
-
-case class ConsulService(name: String, id: String)
-object ConsulService extends DefaultJsonProtocol {
- implicit val serdes = jsonFormat(ConsulService.apply, "Service", "ID")
-}
-
-case class ConsulServiceHealthEntry(service: ConsulService)
-object ConsulServiceHealthEntry extends DefaultJsonProtocol {
- implicit val serdes = jsonFormat(ConsulServiceHealthEntry.apply(_), "Service")
-}
-
-/**
- * Client to access Consul's Health API
- */
-class ConsulHealthApi(base: Uri)(implicit val actorSystem: ActorSystem, val materializer: ActorMaterializer) {
- private implicit val executionContext = actorSystem.dispatcher
-
- /**
- * Gets the health status of the given service
- *
- * @param name name of the service to query for
- * @param passingOnly get only passing nodes of the service
- * @return a future that completes with the matching services
- */
- def service(name: String, passingOnly: Boolean = false): Future[List[ConsulService]] = {
- val healthUri = base.withPath(Uri.Path(s"/v1/health/service/$name"))
- val serviceUri = if (passingOnly) healthUri.withQuery(Uri.Query("passing")) else healthUri
-
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.GET,
- uri = serviceUri))
- r.flatMap { response =>
- Unmarshal(response.entity).to[List[ConsulServiceHealthEntry]]
- } map { _.map(_.service) }
- }
-}
-
-/**
- * Client to access Consul's Catalog API
- */
-class ConsulCatalogApi(base: Uri)(implicit val actorSystem: ActorSystem, val materializer: ActorMaterializer) {
- private implicit val executionContext = actorSystem.dispatcher
-
- /**
- * Gets all services in that consul instance
- *
- * @return a future that completes with a set of service names
- */
- def services(): Future[Set[String]] = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.GET,
- uri = base.withPath(Uri.Path(s"/v1/catalog/services"))))
- r.flatMap { response =>
- Unmarshal(response.entity).to[Map[String, JsValue]]
- } map { _.keys.toSet }
- }
-}
-
-object ConsulClient {
- /**
- * Drops the first level of the keys of the consul entries
- * as it is redundant most of the times.
- *
- * <b>Note:</b> This does not take key differences into account, the
- * caller needs to make sure that no information is lost
- *
- * @param nested a map containing nested keys, e.g.
- * <code>Map("nested/k1" -> "v1", "inner/k2" -> "v2")</code>
- * @return a map with the first level of all keys removed, e.g.
- * <code>Map("k1" -> "v1", "k2" -> "v2")</code>
- */
- def dropKeyLevel(nested: Map[String, String]): Map[String, String] = {
- nested map {
- case (key, value) => key.split("/").tail.mkString("/") -> value
- }
- }
-
- /**
- * Nests a map of consul entries one level by grouping by the
- * first level of the map's keys.
- *
- * @param flat a map with nested keys in a flat manor, e.g.
- * <code>Map("k1/nested" -> "v1", "k2/nested" -> "v2")</code>
- * @return a nested map grouped by the first level of the keys e.g.
- * <code>Map("k1" -> Map("nested" -> "v1"), "k2" -> Map("nested" -> "v2"))</code>
- */
- def toNestedMap(flat: Map[String, String]): Map[String, Map[String, String]] = {
- flat groupBy {
- case (key, value) => key.split("/").head
- } mapValues { dropKeyLevel(_) }
- }
-}
-
-object ConsulKV {
- object InvokerKeys {
- // All invoker written information written here.
- // Underneath this, each invoker has its own path.
- val allInvokers = "invokers" // we store a small amount of data here
- val allInvokersData = "invokersData" // we store large amounts of data here
- private val invokerKeyPrefix = "invoker"
- def instancePath(instance: Int) = s"${allInvokers}/${invokerKeyPrefix}${instance}"
- def instanceDataPath(instance: Int) = s"${allInvokersData}/${invokerKeyPrefix}${instance}"
-
- // Invokers store the hostname they are running on here.
- def hostname(instance: Int) = s"${instancePath(instance)}/hostname"
-
- // Invokers store when they start here.
- val startKey = "start"
- def start(instance: Int) = s"${instancePath(instance)}/$startKey"
-
- // Invokers store their most recent check in time here
- val statusKey = "status"
- def status(instance: Int) = s"${instancePath(instance)}/${statusKey}"
-
- // Extract index from just the element of the path such as "invoker5"
- def extractInvokerIndex(key: String): Int = key.substring(invokerKeyPrefix.length).toInt
-
- // Get the invoker index given a key somewhere in that invoker's KV sub-hierarchy
- def getInvokerIndexFromAny(key: String): Option[Int] = {
- val prefix = s"${allInvokers}/${invokerKeyPrefix}"
- if (key.startsWith(prefix)) {
- val middle = key.substring(prefix.length)
- val slashIndex = middle.indexOf("/")
- if (slashIndex > 0) {
- Try { middle.substring(0, slashIndex).toInt }.toOption
- } else None
- } else None
- }
-
- }
-
- // All keys for information written from the controller are here.
- object ControllerKeys {
- val component = "controller"
- val userActivationCountKey = s"${component}/userActivationCount"
- }
-
- // All load balancer written information under here.
- object LoadBalancerKeys {
- val component = "loadBalancer"
- val hostnameKey = s"${component}/hostname"
- val startKey = s"${component}/start"
- val statusKey = s"${component}/status"
- val activationCountKey = s"${component}/activationCount"
- val overloadKey = s"${component}/overload"
- val invokerHealth = s"${component}/invokerHealth"
- val userActivationCountKey = s"${component}/userActivationCount"
- }
-
- object WhiskProps {
- val whiskProps = "whiskprops"
- }
-}
diff --git a/common/scala/src/main/scala/whisk/common/ConsulKVReporter.scala b/common/scala/src/main/scala/whisk/common/ConsulKVReporter.scala
deleted file mode 100644
index 2fe4cf9..0000000
--- a/common/scala/src/main/scala/whisk/common/ConsulKVReporter.scala
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-package whisk.common
-
-import scala.concurrent.Future
-import scala.concurrent.duration.FiniteDuration
-
-import akka.actor.ActorSystem
-import spray.json.DefaultJsonProtocol.StringJsonFormat
-import spray.json.JsValue
-import spray.json.pimpAny
-
-/**
- * Helper utility to periodically report values to consul's key-value store
- *
- * @param kv instance of the ConsulClient to use
- * @param initialDelay time to wait before starting the reporting initially
- * @param interval time between two reports being send
- * @param hostKey the key for the host address of the component
- * @param startKey the key for the startup timestamp of the component
- * @param statusKey the key for the freshness timestamp of the component
- * @param updater the function to call to update arbitrary values in consul
- */
-class ConsulKVReporter(
- consul: ConsulClient,
- initialDelay: FiniteDuration,
- interval: FiniteDuration,
- hostKey: String,
- startKey: String,
- statusKey: String,
- updater: Int => Map[String, JsValue])(
- implicit val system: ActorSystem,
- logging: Logging) {
-
- implicit val executionContext = system.dispatcher
- private var count = 0
-
- system.scheduler.scheduleOnce(initialDelay) {
- val (selfHostname, _, _) = SimpleExec.syncRunCmd(Array("hostname", "-f"))(TransactionId.unknown, logging)
- consul.kv.put(hostKey, selfHostname.toJson.compactPrint)
- consul.kv.put(startKey, DateUtil.getTimeString.toJson.compactPrint)
-
- Scheduler.scheduleWaitAtLeast(interval) { () =>
- val statusPut = consul.kv.put(statusKey, DateUtil.getTimeString.toJson.compactPrint)
- val updatePuts = updater(count) map {
- case (k, v) => consul.kv.put(k, v.compactPrint)
- }
- count = count + 1
- val allPuts = updatePuts.toSeq :+ statusPut
- Future.sequence(allPuts)
- }
- }
-}
diff --git a/common/scala/src/main/scala/whisk/core/WhiskConfig.scala b/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
index 6af2fcb..8bbe0b4 100644
--- a/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
+++ b/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
@@ -19,15 +19,9 @@ package whisk.core
import java.io.File
-import scala.concurrent.Await
-import scala.concurrent.duration.DurationInt
import scala.io.Source
-import scala.util.Try
-import akka.actor.ActorSystem
import whisk.common.Config
-import whisk.common.ConsulClient
-import whisk.common.ConsulKV
import whisk.common.Logging
/**
@@ -45,7 +39,7 @@ class WhiskConfig(
requiredProperties: Map[String, String],
optionalProperties: Set[String] = Set(),
propertiesFile: File = null,
- env: Map[String, String] = sys.env)(implicit val system: ActorSystem, logging: Logging)
+ env: Map[String, String] = sys.env)(implicit val logging: Logging)
extends Config(requiredProperties, optionalProperties)(env) {
/**
@@ -56,7 +50,6 @@ class WhiskConfig(
override protected def getProperties() = {
val properties = super.getProperties()
WhiskConfig.readPropertiesFromFile(properties, Option(propertiesFile) getOrElse (WhiskConfig.whiskPropertiesFile))
- WhiskConfig.readPropertiesFromConsul(properties)
properties
}
@@ -79,7 +72,6 @@ class WhiskConfig(
val invokerSerializeDockerPull = this(WhiskConfig.invokerSerializeDockerPull)
val invokerUseRunc = this(WhiskConfig.invokerUseRunc)
val invokerUseReactivePool = this(WhiskConfig.invokerUseReactivePool)
- val invokerInstances = this(WhiskConfig.invokerInstances)
val wskApiHost = this(WhiskConfig.wskApiProtocol) + "://" + this(WhiskConfig.wskApiHostname) + ":" + this(WhiskConfig.wskApiPort)
val controllerBlackboxFraction = this.getAsDouble(WhiskConfig.controllerBlackboxFraction, 0.10)
@@ -92,7 +84,6 @@ class WhiskConfig(
val edgeHostName = this(WhiskConfig.edgeHostName)
val zookeeperHost = this(WhiskConfig.zookeeperHostName) + ":" + this(WhiskConfig.zookeeperHostPort)
- val consulServer = this(WhiskConfig.consulServerHost) + ":" + this(WhiskConfig.consulPort)
val invokerHosts = this(WhiskConfig.invokerHostsList)
val dbProvider = this(WhiskConfig.dbProvider)
@@ -142,34 +133,6 @@ object WhiskConfig {
}
/**
- * Reads a Map of key-value pairs from the Consul service -- store them in the
- * mutable properties object.
- */
- def readPropertiesFromConsul(properties: scala.collection.mutable.Map[String, String])(implicit system: ActorSystem, logging: Logging) = {
- //try to get consulServer prop
- val consulString = for {
- server <- properties.get(consulServerHost).filter(s => s != null && s.trim.nonEmpty)
- port <- properties.get(consulPort).filter(_ != null)
- } yield server + ":" + port
-
- consulString match {
- case Some(consulServer) => Try {
- logging.info(this, s"reading properties from consul at $consulServer")
- val consul = new ConsulClient(consulServer)
-
- val whiskProps = Await.result(consul.kv.getRecurse(ConsulKV.WhiskProps.whiskProps), 1.minute)
- properties.keys foreach { p =>
- val kvp = ConsulKV.WhiskProps.whiskProps + "/" + p.replace('.', '_').toUpperCase
- whiskProps.get(kvp) foreach { properties += p -> _ }
- }
- } recover {
- case ex => logging.warn(this, s"failed to read properties from consul: ${ex.getMessage}")
- }
- case _ => logging.info(this, "no consul server defined")
- }
- }
-
- /**
* Reads a Map of key-value pairs from the environment (sys.env) -- store them in the
* mutable properties object.
*/
@@ -237,7 +200,6 @@ object WhiskConfig {
val invokerSerializeDockerPull = "invoker.serializeDockerPull"
val invokerUseRunc = "invoker.useRunc"
val invokerUseReactivePool = "invoker.useReactivePool"
- val invokerInstances = "invoker.instances"
val wskApiProtocol = "whisk.api.host.proto"
val wskApiPort = "whisk.api.host.port"
@@ -260,12 +222,9 @@ object WhiskConfig {
val kafkaHostPort = "kafka.host.port"
private val zookeeperHostPort = "zookeeper.host.port"
- val consulServerHost = "consulserver.host"
- val consulPort = "consul.host.port4"
val invokerHostsList = "invoker.hosts"
val edgeHost = Map(edgeHostName -> null, edgeHostApiPort -> null)
- val consulServer = Map(consulServerHost -> null, consulPort -> null)
val invokerHosts = Map(invokerHostsList -> null)
val kafkaHost = Map(kafkaHostName -> null, kafkaHostPort -> null)
diff --git a/core/controller/src/main/scala/whisk/core/controller/Backend.scala b/core/controller/src/main/scala/whisk/core/controller/Backend.scala
index c09db80..a6c9a54 100644
--- a/core/controller/src/main/scala/whisk/core/controller/Backend.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/Backend.scala
@@ -37,7 +37,4 @@ trait WhiskServices {
/** A load balancing service that launches invocations. */
protected val loadBalancer: LoadBalancer
-
- /** The hostname of the consul server. */
- protected val consulServer: String
}
diff --git a/core/controller/src/main/scala/whisk/core/controller/Controller.scala b/core/controller/src/main/scala/whisk/core/controller/Controller.scala
index bb01a0b..359e78c 100644
--- a/core/controller/src/main/scala/whisk/core/controller/Controller.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/Controller.scala
@@ -118,7 +118,6 @@ class Controller(
// initialize backend services
private implicit val loadBalancer = new LoadBalancerService(whiskConfig, instance, entityStore)
- private implicit val consulServer = whiskConfig.consulServer
private implicit val entitlementProvider = new LocalEntitlementProvider(whiskConfig, loadBalancer)
private implicit val activationIdFactory = new ActivationIdGenerator {}
diff --git a/core/controller/src/main/scala/whisk/core/controller/RestAPIs.scala b/core/controller/src/main/scala/whisk/core/controller/RestAPIs.scala
index 08cf874..2088831 100644
--- a/core/controller/src/main/scala/whisk/core/controller/RestAPIs.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/RestAPIs.scala
@@ -89,7 +89,6 @@ protected[controller] object RestApiCommons {
WhiskAuthStore.requiredProperties ++
WhiskEntityStore.requiredProperties ++
WhiskActivationStore.requiredProperties ++
- WhiskConfig.consulServer ++
EntitlementProvider.requiredProperties ++
WhiskActionsApi.requiredProperties ++
Authenticate.requiredProperties ++
@@ -109,7 +108,6 @@ protected[controller] object RestApiCommons {
override val entitlementProvider: EntitlementProvider,
override val activationIdFactory: ActivationIdGenerator,
override val loadBalancer: LoadBalancerService,
- override val consulServer: String,
override val actorSystem: ActorSystem,
override val executionContext: ExecutionContext,
override val logging: Logging,
@@ -139,7 +137,6 @@ protected[controller] class RestAPIVersion(apipath: String, apiversion: String)(
implicit val entitlementProvider: EntitlementProvider,
implicit val activationIdFactory: ActivationIdGenerator,
implicit val loadBalancer: LoadBalancerService,
- implicit val consulServer: String,
implicit val actorSystem: ActorSystem,
implicit val executionContext: ExecutionContext,
implicit val logging: Logging,
@@ -230,7 +227,6 @@ protected[controller] class RestAPIVersion(apipath: String, apiversion: String)(
override val entitlementProvider: EntitlementProvider,
override val activationIdFactory: ActivationIdGenerator,
override val loadBalancer: LoadBalancerService,
- override val consulServer: String,
override val executionContext: ExecutionContext,
override val logging: Logging,
override val whiskConfig: WhiskConfig)
@@ -248,7 +244,6 @@ protected[controller] class RestAPIVersion(apipath: String, apiversion: String)(
override val activationStore: ActivationStore,
override val activationIdFactory: ActivationIdGenerator,
override val loadBalancer: LoadBalancerService,
- override val consulServer: String,
override val executionContext: ExecutionContext,
override val logging: Logging,
override val whiskConfig: WhiskConfig)
@@ -262,7 +257,6 @@ protected[controller] class RestAPIVersion(apipath: String, apiversion: String)(
override val entitlementProvider: EntitlementProvider,
override val activationIdFactory: ActivationIdGenerator,
override val loadBalancer: LoadBalancerService,
- override val consulServer: String,
override val executionContext: ExecutionContext,
override val logging: Logging,
override val whiskConfig: WhiskConfig)
@@ -284,7 +278,6 @@ protected[controller] class RestAPIVersion(apipath: String, apiversion: String)(
override val entitlementProvider: EntitlementProvider,
override val activationIdFactory: ActivationIdGenerator,
override val loadBalancer: LoadBalancerService,
- override val consulServer: String,
override val executionContext: ExecutionContext,
override val logging: Logging,
override val whiskConfig: WhiskConfig)
diff --git a/core/controller/src/main/scala/whisk/core/entitlement/ActivationThrottler.scala b/core/controller/src/main/scala/whisk/core/entitlement/ActivationThrottler.scala
index f1a80d4..aa44118 100644
--- a/core/controller/src/main/scala/whisk/core/entitlement/ActivationThrottler.scala
+++ b/core/controller/src/main/scala/whisk/core/entitlement/ActivationThrottler.scala
@@ -34,9 +34,11 @@ import whisk.core.loadBalancer.LoadBalancer
* to calculate and determine whether the namespace currently invoking a new action should
* be allowed to do so.
*
- * @param config containing the config information needed (consulServer)
+ * @param loadbalancer contains active quotas
+ * @param defaultConcurrencyLimit the default max allowed concurrent operations
+ * @param systemOverloadLimit the limit when the system is considered overloaded
*/
-class ActivationThrottler(consulServer: String, loadBalancer: LoadBalancer, defaultConcurrencyLimit: Int, systemOverloadLimit: Int)(
+class ActivationThrottler(loadBalancer: LoadBalancer, defaultConcurrencyLimit: Int, systemOverloadLimit: Int)(
implicit val system: ActorSystem, logging: Logging) {
logging.info(this, s"concurrencyLimit = $defaultConcurrencyLimit, systemOverloadLimit = $systemOverloadLimit")
diff --git a/core/controller/src/main/scala/whisk/core/entitlement/Entitlement.scala b/core/controller/src/main/scala/whisk/core/entitlement/Entitlement.scala
index 176ebac..e6b6d19 100644
--- a/core/controller/src/main/scala/whisk/core/entitlement/Entitlement.scala
+++ b/core/controller/src/main/scala/whisk/core/entitlement/Entitlement.scala
@@ -60,7 +60,7 @@ protected[core] case class Resource(
}
protected[core] object EntitlementProvider {
- val requiredProperties = WhiskConfig.consulServer ++ Map(
+ val requiredProperties = Map(
WhiskConfig.actionInvokePerMinuteDefaultLimit -> null,
WhiskConfig.actionInvokeConcurrentDefaultLimit -> null,
WhiskConfig.triggerFirePerMinuteDefaultLimit -> null,
@@ -84,7 +84,7 @@ protected[core] abstract class EntitlementProvider(config: WhiskConfig, loadBala
private val invokeRateThrottler = new RateThrottler("actions per minute", config.actionInvokePerMinuteLimit.toInt, _.limits.invocationsPerMinute)
private val triggerRateThrottler = new RateThrottler("triggers per minute", config.triggerFirePerMinuteLimit.toInt, _.limits.firesPerMinute)
- private val concurrentInvokeThrottler = new ActivationThrottler(config.consulServer, loadBalancer, config.actionInvokeConcurrentLimit.toInt, config.actionInvokeSystemOverloadLimit.toInt)
+ private val concurrentInvokeThrottler = new ActivationThrottler(loadBalancer, config.actionInvokeConcurrentLimit.toInt, config.actionInvokeSystemOverloadLimit.toInt)
/**
* Grants a subject the right to access a resources.
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
index 9868a5e..51598d1 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
@@ -38,12 +38,7 @@ import akka.actor.Props
import akka.pattern.pipe
import akka.util.Timeout
-import spray.json._
-import spray.json.DefaultJsonProtocol._
-
import whisk.common.AkkaLogging
-import whisk.common.ConsulKV.LoadBalancerKeys
-import whisk.common.KeyValueStore
import whisk.common.LoggingMarkers
import whisk.common.RingBuffer
import whisk.common.TransactionId
@@ -82,7 +77,6 @@ final case class InvokerInfo(buffer: RingBuffer[Boolean])
*/
class InvokerPool(
childFactory: (ActorRefFactory, InstanceId) => ActorRef,
- kv: KeyValueStore,
sendActivationToInvoker: (ActivationMessage, InstanceId) => Future[RecordMetadata],
pingConsumer: MessageConsumer) extends Actor {
@@ -123,22 +117,19 @@ class InvokerPool(
refToInstance.get(invoker).foreach { instance =>
status = status.updated(instance.toInt, (instance, currentState))
}
- publishStatus()
+ logStatus()
case Transition(invoker, oldState: InvokerState, newState: InvokerState) =>
refToInstance.get(invoker).foreach {
instance => status = status.updated(instance.toInt, (instance, newState))
}
- publishStatus()
+ logStatus()
// this is only used for the internal test action which enabled an invoker to become healthy again
case msg: ActivationRequest => sendActivationToInvoker(msg.msg, msg.invoker).pipeTo(sender)
}
- def publishStatus() = {
- val json = status.map { case (instance, state) => s"invoker${instance.toInt}" -> state.asString }.toMap.toJson
- kv.put(LoadBalancerKeys.invokerHealth, json.compactPrint)
-
+ def logStatus() = {
val pretty = status.map { case (instance, state) => s"${instance.toInt} -> $state" }
logging.info(this, s"invoker status changed to ${pretty.mkString(", ")}")
}
@@ -169,10 +160,9 @@ class InvokerPool(
object InvokerPool {
def props(
f: (ActorRefFactory, InstanceId) => ActorRef,
- kv: KeyValueStore,
p: (ActivationMessage, InstanceId) => Future[RecordMetadata],
pc: MessageConsumer) = {
- Props(new InvokerPool(f, kv, p, pc))
+ Props(new InvokerPool(f, p, pc))
}
/** A stub identity for invoking the test action. This does not need to be a valid identity. */
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
index 8313832..3418951 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
@@ -35,7 +35,6 @@ import akka.actor.Props
import akka.pattern.ask
import akka.util.Timeout
-import whisk.common.ConsulClient
import whisk.common.Logging
import whisk.common.LoggingMarkers
import whisk.common.TransactionId
@@ -201,12 +200,12 @@ class LoadBalancerService(
}
val maxPingsPerPoll = 128
- val consul = new ConsulClient(config.consulServer)
// Each controller gets its own Group Id, to receive all messages
val pingConsumer = new KafkaConsumerConnector(config.kafkaHost, s"health${instance.toInt}", "health", maxPeek = maxPingsPerPoll)
val invokerFactory = (f: ActorRefFactory, invokerInstance: InstanceId) => f.actorOf(InvokerActor.props(invokerInstance, instance))
- actorSystem.actorOf(InvokerPool.props(invokerFactory, consul.kv,
+ actorSystem.actorOf(InvokerPool.props(
+ invokerFactory,
(m, i) => sendActivationToInvoker(messageProducer, m, i), pingConsumer))
}
@@ -283,8 +282,7 @@ class LoadBalancerService(
}
object LoadBalancerService {
- def requiredProperties = kafkaHost ++ consulServer ++
- Map(loadbalancerInvokerBusyThreshold -> null)
+ def requiredProperties = kafkaHost ++ Map(loadbalancerInvokerBusyThreshold -> null)
/** Memoizes the result of `f` for later use. */
def memoize[I, O](f: I => O): I => O = new scala.collection.mutable.HashMap[I, O]() {
diff --git a/core/invoker/src/main/scala/whisk/core/container/ContainerPool.scala b/core/invoker/src/main/scala/whisk/core/container/ContainerPool.scala
index ee9586d..95e85be 100644
--- a/core/invoker/src/main/scala/whisk/core/container/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/container/ContainerPool.scala
@@ -59,10 +59,6 @@ class ContainerPool(
implicit val executionContext = actorSystem.dispatcher
- // These must be defined before verbosity is set
- private val datastore = WhiskEntityStore.datastore(config)
- private val authStore = WhiskAuthStore.datastore(config)
-
val mounted = !standalone
val dockerhost = config.selfDockerEndpoint
val serializeDockerOp = config.invokerSerializeDockerOp.toBoolean
diff --git a/core/invoker/src/main/scala/whisk/core/container/ContainerUtils.scala b/core/invoker/src/main/scala/whisk/core/container/ContainerUtils.scala
index 85cf555..f00d329 100644
--- a/core/invoker/src/main/scala/whisk/core/container/ContainerUtils.scala
+++ b/core/invoker/src/main/scala/whisk/core/container/ContainerUtils.scala
@@ -92,14 +92,13 @@ trait ContainerUtils {
val cpuArg = Array("-c", cpuShare.toString)
val memoryArg = Array("-m", s"${limits.memory.megabytes}m", "--memory-swap", s"${limits.memory.megabytes}m")
val capabilityArg = Array("--cap-drop", "NET_RAW", "--cap-drop", "NET_ADMIN")
- val consulServiceIgnore = Array("-e", "SERVICE_IGNORE=true")
val fileHandleLimit = Array("--ulimit", "nofile=1024:1024")
val processLimit = Array("--pids-limit", "1024")
val securityOpts = policy map { p => Array("--security-opt", s"apparmor:${p}") } getOrElse (Array.empty[String])
val dnsOpts = dnsServers.map(Seq("--dns", _)).flatten
val containerNetwork = Array("--net", network)
- val cmd = Seq("run") ++ makeEnvVars(env) ++ consulServiceIgnore ++ nameOption ++ cpuArg ++ memoryArg ++
+ val cmd = Seq("run") ++ makeEnvVars(env) ++ nameOption ++ cpuArg ++ memoryArg ++
capabilityArg ++ fileHandleLimit ++ processLimit ++ securityOpts ++ dnsOpts ++ containerNetwork ++ Seq("-d", image) ++ args
runDockerCmd(cmd: _*).toOption.map { result =>
diff --git a/core/invoker/src/main/scala/whisk/core/containerpool/docker/DockerContainer.scala b/core/invoker/src/main/scala/whisk/core/containerpool/docker/DockerContainer.scala
index 869c227..e659937 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/docker/DockerContainer.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/docker/DockerContainer.scala
@@ -72,7 +72,7 @@ object DockerContainer {
implicit docker: DockerApiWithFileAccess, runc: RuncApi, ec: ExecutionContext, log: Logging): Future[DockerContainer] = {
implicit val tid = transid
- val environmentArgs = (environment + ("SERVICE_IGNORE" -> true.toString)).map {
+ val environmentArgs = environment.map {
case (key, value) => Seq("-e", s"$key=$value")
}.flatten
diff --git a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
index 57ba590..dc2f09d 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
@@ -38,7 +38,7 @@ import whisk.common.AkkaLogging
import whisk.common.Scheduler
import whisk.connector.kafka.{ KafkaConsumerConnector, KafkaProducerConnector }
import whisk.core.WhiskConfig
-import whisk.core.WhiskConfig.{ consulServer, dockerImagePrefix, dockerRegistry, kafkaHost, logsDir, servicePort, whiskVersion, invokerUseReactivePool }
+import whisk.core.WhiskConfig.{ dockerImagePrefix, dockerRegistry, kafkaHost, logsDir, servicePort, invokerUseReactivePool }
import whisk.core.connector.{ ActivationMessage, CompletionMessage }
import whisk.core.connector.MessageFeed
import whisk.core.connector.MessageProducer
@@ -424,7 +424,6 @@ class Invoker(
}
private val entityStore = WhiskEntityStore.datastore(config)
- private val authStore = WhiskAuthStore.datastore(config)
private val activationStore = WhiskActivationStore.datastore(config)
private val pool = new ContainerPool(config, instance)
private val activationCounter = new Counter() // global activation counter
@@ -439,16 +438,12 @@ object Invoker {
logsDir -> null,
dockerRegistry -> null,
dockerImagePrefix -> null,
- invokerUseReactivePool -> false.toString,
- WhiskConfig.invokerInstances -> null) ++
+ invokerUseReactivePool -> false.toString) ++
ExecManifest.requiredProperties ++
- WhiskAuthStore.requiredProperties ++
WhiskEntityStore.requiredProperties ++
WhiskActivationStore.requiredProperties ++
ContainerPool.requiredProperties ++
- kafkaHost ++
- consulServer ++
- whiskVersion
+ kafkaHost
def main(args: Array[String]): Unit = {
require(args.length == 1, "invoker instance required")
@@ -504,7 +499,7 @@ object Invoker {
val port = config.servicePort.toInt
BasicHttpService.startService(actorSystem, "invoker", "0.0.0.0", port, new Creator[InvokerServer] {
- def create = new InvokerServer(invokerInstance, config.invokerInstances.toInt)
+ def create = new InvokerServer(invokerInstance, invokerInstance.toInt)
})
}
}
diff --git a/docs/about.md b/docs/about.md
index a757a5e..fa1a78b 100644
--- a/docs/about.md
+++ b/docs/about.md
@@ -18,7 +18,7 @@ An existing catalog of packages offers a quick way to enhance applications with
# How OpenWhisk works
-Being an open-source project, OpenWhisk stands on the shoulders of giants, including Nginx, Kafka, Consul, Docker, CouchDB. All of these components come together to form a “serverless event-based programming service”. To explain all the components in more detail, lets trace an invocation of an action through the system as it happens. An invocation in OpenWhisk is the core thing a serverless-engine does: Execute the code the user has fed into the system and return the results of that execution.
+Being an open-source project, OpenWhisk stands on the shoulders of giants, including Nginx, Kafka, Docker, CouchDB. All of these components come together to form a “serverless event-based programming service”. To explain all the components in more detail, lets trace an invocation of an action through the system as it happens. An invocation in OpenWhisk is the core thing a serverless-engine does: Execute the code the user has fed into the system and return the results of that execution.
## Creating the action
@@ -81,13 +81,9 @@ The record of the action contains mainly the code to execute (shown above) and d
In this particular case, our action doesn’t take any parameters (the function’s parameter definition is an empty list), thus we assume we haven’t set any default parameters and haven’t sent any specific parameters to the action, making for the most trivial case from this point-of-view.
-### Who’s there to invoke the action: Consul
+### Who’s there to invoke the action: Load Balancer
-The Controller (or more specifically the load balancing part of it) has everything in place now to actually get your code running. It needs to know who’s available to do so though. **Consul**, a service discovery, is used to keep track of the executors available in the system by checking their health status continuously. Those executors are called **Invokers**.
-
-The Controller, now knowing which Invokers are available, chooses one of them to invoke the action requested.
-
-Let’s assume for this case, that the system has 3 Invokers available, Invoker 0 to 2, and that the Controller chose *Invoker 2* to invoke the action at hand.
+The Load Balancer, which is part of the Controller, has a global view of the executors available in the system by checking their health status continuously. Those executors are called **Invokers**. The Load Balancer, knowing which Invokers are available, chooses one of them to invoke the action requested.
### Please form a line: Kafka
@@ -98,7 +94,7 @@ From now on, mainly two bad things can happen to the invocation request you sent
The answer to both is **Kafka**, “a high-throughput, distributed, publish-subscribe messaging system”. Controller and Invoker solely communicate through messages buffered and persisted by Kafka. That lifts the burden of buffering in memory, risking an *OutOfMemoryException*, off of both the Controller and the Invoker while also making sure that messages are not lost in case the system crashes.
-To get the action invoked then, the Controller publishes a message to Kafka, which contains the action to invoke and the parameters to pass to that action (in this case none). This message is addressed to the Invoker which the Controller chose above from the list it got from Consul.
+To get the action invoked then, the Controller publishes a message to Kafka, which contains the action to invoke and the parameters to pass to that action (in this case none). This message is addressed to the Invoker which the Controller chose above from the list of available invokers.
Once Kafka has confirmed that it got the message, the HTTP request to the user is responded to with an **ActivationId**. The user will use that later on, to get access to the results of this specific invocation. Note that this is an asynchronous invocation model, where the HTTP request terminates once the system has accepted the request to invoke an action. A synchronous model (called blocking invocation) is available, but not covered by this article.
diff --git a/docs/reference.md b/docs/reference.md
index 2c37ab4..e5ec838 100644
--- a/docs/reference.md
+++ b/docs/reference.md
@@ -456,7 +456,6 @@ The following table lists the default limits for actions.
### Per namespace concurrent invocation (Default: 100)
* The number of activations that are either executing or queued for execution for a namespace cannot exceed 100.
-* The default limit can be statically configured by whisk in consul kvstore.
* A user is currently not able to change the limits.
### Invocations per minute (Fixed: 120)
diff --git a/tests/src/test/scala/common/WhiskProperties.java b/tests/src/test/scala/common/WhiskProperties.java
index 70176e7..32ab0b0 100644
--- a/tests/src/test/scala/common/WhiskProperties.java
+++ b/tests/src/test/scala/common/WhiskProperties.java
@@ -144,14 +144,6 @@ public class WhiskProperties {
return Integer.parseInt(whiskProperties.getProperty("kafkaras.host.port"));
}
- public static String getConsulServerHost() {
- return whiskProperties.getProperty("consulserver.host");
- }
-
- public static int getConsulKVPort() {
- return Integer.parseInt(whiskProperties.getProperty("consul.host.port4"));
- }
-
public static String getZookeeperHost() {
return whiskProperties.getProperty("zookeeper.host");
}
diff --git a/tests/src/test/scala/whisk/consul/ConsulClientTests.scala b/tests/src/test/scala/whisk/consul/ConsulClientTests.scala
deleted file mode 100644
index 8084453..0000000
--- a/tests/src/test/scala/whisk/consul/ConsulClientTests.scala
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-package whisk.consul
-
-import scala.concurrent.duration.DurationInt
-
-import org.junit.runner.RunWith
-import org.scalatest.FlatSpec
-import org.scalatest.Matchers
-import org.scalatest.concurrent.ScalaFutures
-import org.scalatest.junit.JUnitRunner
-import org.scalatest.time.Span.convertDurationToSpan
-
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
-import akka.http.scaladsl.marshalling._
-import akka.http.scaladsl.model._
-import akka.http.scaladsl.unmarshalling._
-import akka.stream.ActorMaterializer
-import common.StreamLogging
-import common.WskActorSystem
-import spray.json._
-import spray.json.DefaultJsonProtocol._
-import whisk.common.ConsulClient
-import whisk.common.ConsulService
-import whisk.core.WhiskConfig
-import whisk.core.WhiskConfig.consulServer
-import whisk.utils.retry
-
-@RunWith(classOf[JUnitRunner])
-class ConsulClientTests
- extends FlatSpec
- with ScalaFutures
- with Matchers
- with WskActorSystem
- with StreamLogging {
-
- implicit val testConfig = PatienceConfig(5.seconds)
- implicit val materializer = ActorMaterializer()
-
- val config = new WhiskConfig(consulServer)
- val consul = new ConsulClient(config.consulServer)
-
- val consulHost :: consulPort :: Nil = config.consulServer.split(":").toList
- val consulUri = Uri().withScheme("http").withHost(consulHost).withPort(consulPort.toInt)
- val checkInterval = 1.second
-
- /**
- * Registers a service in consul with a given healthcheck
- */
- def registerService(name: String, id: String, checkScript: Option[String] = None) = {
- val obj = Map(
- "ID" -> id.toJson,
- "Name" -> name.toJson) ++ checkScript.map { script =>
- "Check" -> JsObject(
- "Script" -> script.toJson,
- "Interval" -> s"${checkInterval.toSeconds}s".toJson)
- }
-
- Marshal(obj.toJson).to[RequestEntity].flatMap { entity =>
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.PUT,
- uri = consulUri.withPath(Uri.Path("/v1/agent/service/register")),
- entity = entity))
- r.flatMap { response =>
- Unmarshal(response).to[Any]
- }
- }
- }
-
- /**
- * Deregisters a service in consul
- */
- def deregisterService(id: String) = {
- val r = Http().singleRequest(
- HttpRequest(
- method = HttpMethods.PUT,
- uri = consulUri.withPath(Uri.Path(s"/v1/agent/service/deregister/$id"))))
- r.flatMap { response =>
- Unmarshal(response).to[Any]
- }
- }
-
- "Consul KV client" should "be able to put, get and delete" in {
- val key = "emperor"
- val value = "palpatine"
-
- // Create an entry
- noException should be thrownBy consul.kv.put(key, value).futureValue
-
- // Gets the entry
- consul.kv.get(key).futureValue should equal(value)
-
- // Deletes the entry
- noException should be thrownBy consul.kv.del(key).futureValue
-
- // Asserts that the entry is gone
- consul.kv.get(key).failed.futureValue shouldBe a[NoSuchElementException]
- }
-
- it should "return an Exception if a non-existent key is queried" in {
- val key = "no_such_key"
- consul.kv.get(key).failed.futureValue shouldBe a[NoSuchElementException]
- }
-
- it should "be able to retrieve many keys recursively and checking against stored values" in {
- val prefix = "xxx"
- val values = Map(
- s"$prefix/k1" -> "key1",
- s"$prefix/k2" -> "key2",
- s"$prefix/k3/inner" -> "key3")
-
- // Write all values to the key/value store
- values foreach {
- case (key, value) => noException should be thrownBy consul.kv.put(key, value).futureValue
- }
-
- consul.kv.getRecurse(prefix).futureValue should equal(values)
- }
-
- "Consul Catalog client" should "return a list of all services" in {
- consul.catalog.services().futureValue.size should be > 0
- }
-
- "Consul Health client" should "return an empty list of health results" in {
- val services = consul.health.service("bogus").futureValue
- services shouldBe List.empty
- }
-
- it should "return a list of health results" in {
- val service = ConsulService("testservice", "testservice_1")
- registerService(service.name, service.id).futureValue
-
- val services = consul.health.service(service.name).futureValue
- // Immediately deregister before actually asserting to clean up
- deregisterService(service.id).futureValue
-
- services.head shouldBe service
- }
-
- it should "return only the passing service" in {
- val passing = ConsulService("testservice", "testservice_passing")
- val failing = ConsulService("testservice", "testservice_failing")
- registerService(passing.name, passing.id, Some("exit 0")).futureValue
- registerService(failing.name, failing.id, Some("exit 1")).futureValue
-
- try {
- retry(consul.health.service(passing.name, true).futureValue.head shouldBe passing, 3, Some(checkInterval))
- } finally {
- // Make sure, that deregister runs, even if test fails
- deregisterService(passing.id).futureValue
- deregisterService(failing.id).futureValue
- }
- }
-
- "ConsulClient helper methods" should "drop the first part of the key" in {
- val test = Map(
- "nested/k1" -> "v1",
- "inner/k2" -> "v2")
-
- val flattened = Map(
- "k1" -> "v1",
- "k2" -> "v2")
-
- ConsulClient.dropKeyLevel(test) should equal(flattened)
- }
-
- it should "create a nested map by grouping by the first part of the key" in {
- val test = Map(
- "invoker0/count" -> "1",
- "invoker0/status" -> "true",
- "invoker1/count" -> "2",
- "invoker1/status" -> "false")
-
- val nested = Map(
- "invoker0" -> Map("count" -> "1", "status" -> "true"),
- "invoker1" -> Map("count" -> "2", "status" -> "false"))
-
- ConsulClient.toNestedMap(test) should equal(nested)
- }
-
-}
diff --git a/tests/src/test/scala/whisk/consul/ConsulHealthTests.scala b/tests/src/test/scala/whisk/consul/ConsulHealthTests.scala
deleted file mode 100644
index f5e466b..0000000
--- a/tests/src/test/scala/whisk/consul/ConsulHealthTests.scala
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-package whisk.consul
-
-import scala.concurrent.Future
-import scala.concurrent.duration.DurationInt
-
-import org.junit.runner.RunWith
-import org.scalatest.FlatSpec
-import org.scalatest.Matchers
-import org.scalatest.concurrent.ScalaFutures
-import org.scalatest.junit.JUnitRunner
-import org.scalatest.time.Span.convertDurationToSpan
-
-import common.StreamLogging
-import common.WskActorSystem
-import whisk.common.ConsulClient
-import whisk.core.WhiskConfig
-import whisk.core.WhiskConfig.consulServer
-
-@RunWith(classOf[JUnitRunner])
-class ConsulHealthTests
- extends FlatSpec
- with ScalaFutures
- with Matchers
- with WskActorSystem
- with StreamLogging {
-
- implicit val testConfig = PatienceConfig(5.seconds)
-
- private val config = new WhiskConfig(consulServer)
- private val consul = new ConsulClient(config.consulServer)
-
- "Consul" should "have all components passing" in {
- val services = consul.catalog.services().futureValue
-
- val health = services.map { service => consul.health.service(service) }.toList
- val healthResults = Future.sequence(health).futureValue
-
- val filteredHealth = services.map { service => consul.health.service(service, true) }.toList
- val filteredHealthResults = Future.sequence(filteredHealth).futureValue
-
- healthResults should contain theSameElementsAs filteredHealthResults
- }
-
-}
diff --git a/tests/src/test/scala/whisk/core/WhiskConfigTests.scala b/tests/src/test/scala/whisk/core/WhiskConfigTests.scala
index da1a365..a00e087 100644
--- a/tests/src/test/scala/whisk/core/WhiskConfigTests.scala
+++ b/tests/src/test/scala/whisk/core/WhiskConfigTests.scala
@@ -21,23 +21,17 @@ import java.io.BufferedWriter
import java.io.File
import java.io.FileWriter
-import scala.concurrent.Await
-import scala.concurrent.duration._
-
import org.junit.runner.RunWith
import org.scalatest.FlatSpec
import org.scalatest.Matchers
import org.scalatest.junit.JUnitRunner
import common.StreamLogging
-import common.WskActorSystem
-import whisk.common.ConsulClient
@RunWith(classOf[JUnitRunner])
class WhiskConfigTests
extends FlatSpec
with Matchers
- with WskActorSystem
with StreamLogging {
behavior of "WhiskConfig"
@@ -97,21 +91,4 @@ class WhiskConfigTests
println(s"${WhiskConfig.dockerRegistry} is: '${config.dockerRegistry}'")
assert(config.isValid)
}
-
- it should "read properties from consulserver" in {
- val tester = new WhiskConfig(WhiskConfig.consulServer);
- val consul = new ConsulClient(tester.consulServer)
-
- val key = "whiskprops/CONSUL_TEST_CASE"
- Await.result(consul.kv.put(key, "thiswastested"), 10.seconds)
-
- // set optional value which will not be available in environment, it should still be read from consul
- val config = new WhiskConfig(WhiskConfig.consulServer ++ Map("consul.test.case" -> null), Set("consul.test.case"))
-
- assert(config.isValid)
- assert(config("consul.test.case").equals("thiswastested"))
-
- consul.kv.del(key)
- }
-
}
diff --git a/tests/src/test/scala/whisk/core/containerpool/docker/test/DockerContainerTests.scala b/tests/src/test/scala/whisk/core/containerpool/docker/test/DockerContainerTests.scala
index f892a4b..67dfa33 100644
--- a/tests/src/test/scala/whisk/core/containerpool/docker/test/DockerContainerTests.scala
+++ b/tests/src/test/scala/whisk/core/containerpool/docker/test/DockerContainerTests.scala
@@ -144,7 +144,7 @@ class DockerContainerTests extends FlatSpec
args should contain inOrder ("--name", name)
// Assert proper environment passing
- args should contain allOf ("-e", "test=hi", "SERVICE_IGNORE=true")
+ args should contain allOf ("-e", "test=hi")
}
it should "pull a user provided image before creating the container" in {
diff --git a/tests/src/test/scala/whisk/core/controller/test/ControllerTestCommon.scala b/tests/src/test/scala/whisk/core/controller/test/ControllerTestCommon.scala
index 568e434..49864b9 100644
--- a/tests/src/test/scala/whisk/core/controller/test/ControllerTestCommon.scala
+++ b/tests/src/test/scala/whisk/core/controller/test/ControllerTestCommon.scala
@@ -84,8 +84,6 @@ protected trait ControllerTestCommon
override def make = fixedId
}
- override val consulServer = "???"
-
val entityStore = WhiskEntityStore.datastore(whiskConfig)
val activationStore = WhiskActivationStore.datastore(whiskConfig)
val authStore = WhiskAuthStore.datastore(whiskConfig)
diff --git a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
index 371fad2..50db9a6 100644
--- a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
+++ b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
@@ -45,8 +45,6 @@ import akka.testkit.TestKit
import akka.testkit.TestProbe
import akka.util.Timeout
import common.StreamLogging
-import whisk.common.ConsulKV.LoadBalancerKeys
-import whisk.common.KeyValueStore
import whisk.common.TransactionId
import whisk.core.WhiskConfig
import whisk.core.connector.ActivationMessage
@@ -118,10 +116,8 @@ class InvokerSupervisionTests extends TestKit(ActorSystem("InvokerSupervision"))
val children = mutable.Queue(invoker5.ref, invoker2.ref)
val childFactory = (f: ActorRefFactory, instance: InstanceId) => children.dequeue()
- val kv = stub[KeyValueStore]
val sendActivationToInvoker = stubFunction[ActivationMessage, InstanceId, Future[RecordMetadata]]
-
- val supervisor = system.actorOf(InvokerPool.props(childFactory, kv, sendActivationToInvoker, pC))
+ val supervisor = system.actorOf(InvokerPool.props(childFactory, sendActivationToInvoker, pC))
within(timeout.duration) {
// create first invoker
@@ -154,45 +150,14 @@ class InvokerSupervisionTests extends TestKit(ActorSystem("InvokerSupervision"))
}
}
- it should "publish state changes via kv and call the provided callback if an invoker goes offline" in {
- val invoker = TestProbe()
- val invokerInstance = InstanceId(0)
- val invokerName = s"invoker${invokerInstance.toInt}"
- val childFactory = (f: ActorRefFactory, instance: InstanceId) => invoker.ref
-
- val kv = stub[KeyValueStore]
- val sendActivationToInvoker = stubFunction[ActivationMessage, InstanceId, Future[RecordMetadata]]
- val supervisor = system.actorOf(InvokerPool.props(childFactory, kv, sendActivationToInvoker, pC))
-
- within(timeout.duration) {
- // create first invoker
- val ping0 = PingMessage(invokerInstance)
- supervisor ! ping0
- invoker.expectMsgType[SubscribeTransitionCallBack] // subscribe to the actor
- invoker.expectMsg(ping0)
-
- // triggers kv.put
- invoker.send(supervisor, CurrentState(invoker.ref, Healthy))
- // triggers kv.put and callback
- invoker.send(supervisor, Transition(invoker.ref, Healthy, Offline))
- // triggers another kv.put
- invoker.send(supervisor, Transition(invoker.ref, Offline, Healthy))
- }
-
- retry({
- (kv.put _).verify(LoadBalancerKeys.invokerHealth, *).repeated(3)
- }, N = 3, waitBeforeRetry = Some(500.milliseconds))
- }
-
it should "forward the ActivationResult to the appropriate invoker" in {
val invoker = TestProbe()
val invokerInstance = InstanceId(0)
val invokerName = s"invoker${invokerInstance.toInt}"
val childFactory = (f: ActorRefFactory, instance: InstanceId) => invoker.ref
- val kv = stub[KeyValueStore]
val sendActivationToInvoker = stubFunction[ActivationMessage, InstanceId, Future[RecordMetadata]]
- val supervisor = system.actorOf(InvokerPool.props(childFactory, kv, sendActivationToInvoker, pC))
+ val supervisor = system.actorOf(InvokerPool.props(childFactory, sendActivationToInvoker, pC))
within(timeout.duration) {
// Create one invoker
@@ -216,10 +181,9 @@ class InvokerSupervisionTests extends TestKit(ActorSystem("InvokerSupervision"))
val invokerName = s"invoker${invokerInstance.toInt}"
val childFactory = (f: ActorRefFactory, instance: InstanceId) => invoker.ref
- val kv = stub[KeyValueStore]
val sendActivationToInvoker = stubFunction[ActivationMessage, InstanceId, Future[RecordMetadata]]
- val supervisor = system.actorOf(InvokerPool.props(childFactory, kv, sendActivationToInvoker, pC))
+ val supervisor = system.actorOf(InvokerPool.props(childFactory, sendActivationToInvoker, pC))
// Send ActivationMessage to InvokerPool
val activationMessage = ActivationMessage(
diff --git a/tools/build/redo b/tools/build/redo
index 9ffe233..f47c831 100755
--- a/tools/build/redo
+++ b/tools/build/redo
@@ -239,10 +239,6 @@ Components = [
'teardown all deployed containers',
yaml = 'teardown.yml'),
- makeComponent('consul',
- 'build/deploy consul (includes registrator)',
- modes = 'clean'),
-
makeComponent('kafka',
'build/deploy kafka',
modes = 'clean'),
diff --git a/tools/health/isAlive b/tools/health/isAlive
index 86c39f8..cd445c9 100755
--- a/tools/health/isAlive
+++ b/tools/health/isAlive
@@ -85,18 +85,6 @@ def isElkAlive() :
okString = '"cluster_name"'
return monitorUtil.outputChecker(cmd, okString, 1)
-def isConsulAlive() :
- host = getProp('CONSULSERVER_HOST')
- port = getProp('CONSUL_HOST_PORT4')
- cmd = ['curl', '-m', '3', '-s', '-X', 'PUT', '-d', 'true', '%s:%s/v1/kv/consulIsAlive' % (host, port)]
- okString = 'true'
- rc = monitorUtil.outputChecker(cmd, okString, 1)
- if rc == 0:
- #delete temp property again
- cmd = ['curl', '-m', '3', '-s', '-X', 'DELETE', '%s:%s/v1/kv/consulIsAlive' % (host, port)]
- monitorUtil.run(cmd)
- return rc
-
def isZookeeperAlive() :
host_port = getHostAndPort('zookeeper').split(':')
cmd = '(echo ruok; sleep 1) | nc '+ host_port[0] + ' '+ host_port[1]
@@ -137,8 +125,6 @@ elif (args.component == 'elk'):
monitorUtil.checkLoop(args.component, isElkAlive, delay)
elif (args.component == 'zookeeper'):
monitorUtil.checkLoop(args.component, isZookeeperAlive, delay)
-elif (args.component == 'consulserver'):
- monitorUtil.checkLoop(args.component, isConsulAlive, delay)
else:
host_port = getHostAndPort(args.component, args.host)
monitorUtil.checkLoop(args.component, lambda : ping(host_port), delay)
diff --git a/tools/health/kvstore b/tools/health/kvstore
deleted file mode 100755
index cc991a0..0000000
--- a/tools/health/kvstore
+++ /dev/null
@@ -1,144 +0,0 @@
-#!/usr/bin/env python
-
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-#
-
-#
-# >>>>> If this script is moved, the path manipulation needs to be adjusted. See below. <<<<<
-#
-# Usage:
-# kvstore dump the kv store once
-# kvstore --repeat dump the kv store repeatedly with 3 seconds in between
-# kvstore --import -d dir import whisk.properties into consul
-
-# Start with standard imports.
-import os
-import sys
-import argparse
-import time
-import json
-import base64
-
-# ************************************************************************************
-# * Path stuff. If this script is moved, it should be enough to adjust rootDir only. *
-# ************************************************************************************
-scriptDir = sys.path[0]
-rootDir = os.path.dirname(os.path.dirname(scriptDir))
-cliDir = os.path.join(rootDir, "tools", "admin")
-sys.path.insert(1, cliDir)
-
-# After path adjusted, do the remaining imports.
-import wskprop
-from monitorUtil import run
-
-# Get the substring before the first slash if present else the whole string
-def getFirst(str):
- pos = str.find("/")
- return str if pos<0 else str[0:pos]
-
-# Make it look nice by grouping
-def layerRecords(records):
- cur = 0
- curKey = ''
- while (cur < len(records)):
- nextKey = getFirst(records[cur]['Key'])
- if (curKey != nextKey):
- curKey = nextKey
- records.insert(cur, { 'Key' : curKey, 'Value' : '' })
- else:
- records[cur]['Key'] = ' ' + records[cur]['Key'][len(curKey)+1:]
- cur = cur+1
-
-# Base 64 decoding is used for bulk get so we decode the 'Value' field
-def processRecords(records, interactive):
- for rec in records:
- rawVal = rec['Value']
- if (rawVal is not None):
- rec['Value'] = base64.b64decode(rawVal)
- def getKey(rec):
- rec['Value']
- sorted(records, key=getKey)
- if (interactive):
- layerRecords(records)
-
-# Print all key-value pairs in the store
-def dump(interactive,whiskprops):
- consulserverHost = getProp(whiskprops,'CONSULSERVER_HOST')
- consulHostPort4 = getProp(whiskprops,'CONSUL_HOST_PORT4')
- # Fetch the data and process it
- url = 'http://' + consulserverHost + ':' + consulHostPort4 + '/v1/kv/?recurse'
- getCmd = ['curl', '-s', url]
- (rc, output) = run(getCmd)
- records = json.loads(output) # 'Key', 'Value' are 2 of the fields
- processRecords(records, interactive)
- # Perform actual display
- if (interactive):
- sys.stderr.write("\x1b[2J\x1b[H")
- print(time.strftime("%Y/%m/%d %H:%M:%S"))
- print('-------------------')
- for rec in records:
- print('{0:50} {1}'.format(rec['Key'], rec['Value']))
-
-# write whiskprops to kvstore
-def importProps(whiskprops):
- # We grab all the properties out of whiskprops here to make dependence clear.
- consulserverHost = getProp(whiskprops,'CONSULSERVER_HOST')
- consulHostPort4 = getProp(whiskprops,'CONSUL_HOST_PORT4')
-
- for key in whiskprops:
- url = 'http://' + consulserverHost + ':' + consulHostPort4 + '/v1/kv/whiskprops/%s' % key
- cmd = ['curl', '-X', 'PUT', '-H', 'Content-Type: text/plain',
- '-d', '%s' % whiskprops[key], url]
- (rc, output) = run(cmd)
-
-def getProp(whiskprops, key, defaultValue = None) :
- try:
- return whiskprops[key]
- except KeyError:
- if (defaultValue == None):
- print('Could not find %s' % key)
- return defaultValue
-
-
-parser = argparse.ArgumentParser(description='Read or write to whisk key-value store')
-parser.add_argument('--repeat', help='query the store, repeat forever')
-parser.add_argument('--import', dest='imp', help='host', action='store_true')
-parser.add_argument('-d','--dir', help='directory for whisk.properties')
-args = parser.parse_args()
-
-if (args.dir is None):
- print('Must specify directory with -d')
- exit(-1)
-
-# Use wskprop to read in whisk.properties and provide our own lookup
-whiskprops = wskprop.importPropsIfAvailable(wskprop.propfile(args.dir))
-
-if (args.repeat):
- try:
- while True:
- dump(True,whiskprops)
- time.sleep(1)
- except KeyboardInterrupt:
- True # catch but do nothing so ctrl-c does cause a stack trace
-else:
- if (args.imp):
-
- print("importing whisk.properties into consul kv...")
- importProps(whiskprops)
- print("done.")
- else:
- dump(False,whiskprops)
--
To stop receiving notification emails like this one, please contact
['"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>'].