move away require libv2/config from libv2/redis

Use a list of Redis sentinels that are running on stateful nodes only,
instead of localhost.

Previously, a stateless-only node wouldn't have a local sentinel node
running, causing UtapiReindex to fail.

Added a failover mechanism in case of connection error on the current
sentinel, to try each other one in turn.

.eslintrc

@@ -1,8 +1,18 @@
 {
   "extends": "scality",
+  "env": {
+    "es6": true
+  },
+  "parserOptions": {
+    "ecmaVersion": 9
+  },
   "rules": {
     "no-underscore-dangle": "off",
-	  "implicit-arrow-linebreak" : "off"
+	  "implicit-arrow-linebreak" : "off",
+    "import/extensions": 0,
+    "prefer-spread": 0,
+    "no-param-reassign": 0,
+    "array-callback-return": 0
   },
   "settings": {
       "import/resolver": {

.github/ISSUE_TEMPLATE.md

@@ -0,0 +1,87 @@
+# General support information
+
+GitHub Issues are **reserved** for actionable bug reports (including
+documentation inaccuracies), and feature requests.
+**All questions** (regarding configuration, usecases, performance, community,
+events, setup and usage recommendations, among other things) should be asked on
+the **[Zenko Forum](http://forum.zenko.io/)**.
+
+> Questions opened as GitHub issues will systematically be closed, and moved to
+> the [Zenko Forum](http://forum.zenko.io/).
+
+--------------------------------------------------------------------------------
+
+## Avoiding duplicates
+
+When reporting a new issue/requesting a feature, make sure that we do not have
+any duplicates already open:
+
+- search the issue list for this repository (use the search bar, select
+  "Issues" on the left pane after searching);
+- if there is a duplicate, please do not open your issue, and add a comment
+  to the existing issue instead.
+
+--------------------------------------------------------------------------------
+
+## Bug report information
+
+(delete this section (everything between the lines) if you're not reporting a
+bug but requesting a feature)
+
+### Description
+
+Briefly describe the problem you are having in a few paragraphs.
+
+### Steps to reproduce the issue
+
+Please provide steps to reproduce, including full log output
+
+### Actual result
+
+Describe the results you received
+
+### Expected result
+
+Describe the results you expected
+
+### Additional information
+
+- Node.js version,
+- Docker version,
+- npm version,
+- distribution/OS,
+- optional: anything else you deem helpful to us.
+
+--------------------------------------------------------------------------------
+
+## Feature Request
+
+(delete this section (everything between the lines) if you're not requesting
+a feature but reporting a bug)
+
+### Proposal
+
+Describe the feature
+
+### Current behavior
+
+What currently happens
+
+### Desired behavior
+
+What you would like to happen
+
+### Usecase
+
+Please provide usecases for changing the current behavior
+
+### Additional information
+
+- Is this request for your company? Y/N
+  - If Y: Company name:
+  - Are you using any Scality Enterprise Edition products (RING, Zenko EE)? Y/N
+- Are you willing to contribute this feature yourself?
+- Position/Title:
+- How did you hear about us?
+
+--------------------------------------------------------------------------------

.github/docker/redis-replica/Dockerfile

@@ -0,0 +1,14 @@
+# Creating this image for the CI as GitHub Actions
+# is unable to overwrite the entrypoint
+
+ARG REDIS_IMAGE="redis:latest"
+FROM ${REDIS_IMAGE}
+
+ENV REDIS_LISTEN_PORT 6380
+ENV REDIS_MASTER_HOST redis
+ENV REDIS_MASTER_PORT_NUMBER 6379
+
+
+ENTRYPOINT redis-server \
+    --port ${REDIS_LISTEN_PORT} \
+    --slaveof ${REDIS_MASTER_HOST} ${REDIS_MASTER_PORT_NUMBER}

.github/docker/vault/Dockerfile

@@ -0,0 +1,7 @@
+FROM ghcr.io/scality/vault:c2607856
+
+ENV VAULT_DB_BACKEND LEVELDB
+
+RUN chmod 400 tests/utils/keyfile
+
+ENTRYPOINT yarn start

.github/scripts/run_ft_tests.bash

@@ -17,6 +17,6 @@ if [ -z "$SETUP_CMD" ]; then
   SETUP_CMD="start"
 fi
 
-UTAPI_INTERVAL_TEST_MODE=$1 npm $SETUP_CMD 2>&1 | tee -a "/artifacts/setup_$2.log" &
+UTAPI_INTERVAL_TEST_MODE=$1 npm $SETUP_CMD 2>&1 | tee -a "setup_$2.log" &
 bash tests/utils/wait_for_local_port.bash $PORT 40
-UTAPI_INTERVAL_TEST_MODE=$1 npm run $2 | tee -a "/artifacts/test_$2.log"
+UTAPI_INTERVAL_TEST_MODE=$1 npm run $2 | tee -a "test_$2.log"

.github/workflows/build-ci.yaml

@@ -0,0 +1,65 @@
+name: build-ci-images
+
+on:
+  workflow_call:
+
+jobs:
+  warp10-ci:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    secrets:
+      REGISTRY_LOGIN: ${{ github.repository_owner }}
+      REGISTRY_PASSWORD: ${{ github.token }}
+    with:
+      name: warp10-ci
+      context: .
+      file: images/warp10/Dockerfile
+      lfs: true
+
+  redis-ci:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    secrets:
+      REGISTRY_LOGIN: ${{ github.repository_owner }}
+      REGISTRY_PASSWORD: ${{ github.token }}
+    with:
+      name: redis-ci
+      context: .
+      file: images/redis/Dockerfile
+
+  redis-replica-ci:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    needs:
+      - redis-ci
+    secrets:
+      REGISTRY_LOGIN: ${{ github.repository_owner }}
+      REGISTRY_PASSWORD: ${{ github.token }}
+    with:
+      name: redis-replica-ci
+      context: .github/docker/redis-replica
+      build-args: |
+        REDIS_IMAGE=ghcr.io/${{ github.repository }}/redis-ci:${{ github.sha }}
+
+  vault-ci:
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          lfs: true
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+      - name: Login to GitHub Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ github.token }}
+
+      - name: Build and push vault Image
+        uses: docker/build-push-action@v5
+        with:
+          push: true
+          context: .github/docker/vault
+          tags: ghcr.io/${{ github.repository }}/vault-ci:${{ github.sha }}
+          cache-from: type=gha,scope=vault
+          cache-to: type=gha,mode=max,scope=vault

.github/workflows/build-dev.yaml

@@ -0,0 +1,16 @@
+name: build-dev-image
+
+on:
+  push:
+    branches-ignore:
+    - 'development/**'
+
+jobs:
+  build-dev:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    secrets:
+      REGISTRY_LOGIN: ${{ github.repository_owner }}
+      REGISTRY_PASSWORD: ${{ github.token }}
+    with:
+      namespace: ${{ github.repository_owner }}
+      name: ${{ github.event.repository.name }}

.github/workflows/release-warp10.yaml

@@ -0,0 +1,39 @@
+name: release-warp10
+
+on:
+  workflow_dispatch:
+    inputs:
+      tag:
+        type: string
+        description: 'Tag to be released'
+        required: true
+      create-github-release:
+        type: boolean
+        description: Create a tag and matching Github release.
+        required: false
+        default: true
+
+jobs:
+  build:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    secrets: inherit
+    with:
+      name: warp10
+      context: .
+      file: images/warp10/Dockerfile
+      tag: ${{ github.event.inputs.tag }}
+      lfs: true
+
+  release:
+    if: ${{ inputs.create-github-release }}
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: softprops/action-gh-release@v2
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          name: Release utapi/warp10:${{ github.event.inputs.tag }}-warp10
+          tag_name: ${{ github.event.inputs.tag }}-warp10
+          generate_release_notes: false
+          target_commitish: ${{ github.sha }}

.github/workflows/release.yaml

@@ -0,0 +1,45 @@
+name: release
+
+on:
+  workflow_dispatch:
+    inputs:
+      dockerfile:
+        description: Dockerfile to build image from
+        type: choice
+        options:
+          - images/nodesvc-base/Dockerfile
+          - Dockerfile
+        required: true
+      tag:
+        type: string
+        description: 'Tag to be released'
+        required: true
+      create-github-release:
+        type: boolean
+        description: Create a tag and matching Github release.
+        required: false
+        default: false
+
+jobs:
+  build:
+    uses: scality/workflows/.github/workflows/docker-build.yaml@v2
+    with:
+      namespace: ${{ github.repository_owner }}
+      name: ${{ github.event.repository.name }}
+      context: .
+      file: ${{ github.event.inputs.dockerfile}}
+      tag: ${{ github.event.inputs.tag }}
+
+  release:
+    if: ${{ inputs.create-github-release }}
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: softprops/action-gh-release@v2
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        with:
+          name: Release ${{ github.event.inputs.tag }}
+          tag_name: ${{ github.event.inputs.tag }}
+          generate_release_notes: true
+          target_commitish: ${{ github.sha }}

.github/workflows/tests.yaml

@@ -0,0 +1,361 @@
+---
+name: tests
+
+on:
+  push:
+    branches-ignore:
+      - 'development/**'
+
+  workflow_dispatch:
+    inputs:
+      debug:
+        description: Debug (enable the ability to SSH to runners)
+        type: boolean
+        required: false
+        default: 'false'
+      connection-timeout-m:
+        type: number
+        required: false
+        description: Timeout for ssh connection to worker (minutes)
+        default: 30
+jobs:
+  build-ci:
+    uses: ./.github/workflows/build-ci.yaml
+
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        lfs: true
+    - uses: actions/setup-node@v4
+      with:
+        node-version: '16.13.2'
+        cache: yarn
+    - name: install dependencies
+      run: yarn install --frozen-lockfile --network-concurrency 1
+    - name: run static analysis tools on markdown
+      run: yarn run lint_md
+    - name: run static analysis tools on code
+      run: yarn run lint
+
+  tests-v1:
+    needs:
+      - build-ci
+    runs-on: ubuntu-latest
+    env:
+      REINDEX_PYTHON_INTERPRETER: python3
+    name: ${{ matrix.test.name }}
+    strategy:
+      fail-fast: false
+      matrix:
+        test:
+        - name: run unit tests
+          command: yarn test
+          env:
+            UTAPI_METRICS_ENABLED: 'true'
+        - name: run v1 client tests
+          command: bash ./.github/scripts/run_ft_tests.bash false ft_test:client
+          env: {}
+        - name: run v1 server tests
+          command: bash ./.github/scripts/run_ft_tests.bash false ft_test:server
+          env: {}
+        - name: run v1 cron tests
+          command: bash ./.github/scripts/run_ft_tests.bash false ft_test:cron
+          env: {}
+        - name: run v1 interval tests
+          command: bash ./.github/scripts/run_ft_tests.bash true ft_test:interval
+          env: {}
+    services:
+      redis:
+        image: ghcr.io/${{ github.repository }}/redis-ci:${{ github.sha }}
+        ports:
+        - 6379:6379
+        - 9121:9121
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-replica:
+        image: ghcr.io/${{ github.repository }}/redis-replica-ci:${{ github.sha }}
+        ports:
+        - 6380:6380
+        options: >-
+          --health-cmd "redis-cli -p 6380 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-sentinel:
+        image: bitnami/redis-sentinel:7.2.4
+        env:
+          REDIS_MASTER_SET: scality-s3
+          REDIS_SENTINEL_PORT_NUMBER: '16379'
+          REDIS_SENTINEL_QUORUM: '1'
+        ports:
+        - 16379:16379
+        options: >-
+          --health-cmd "redis-cli -p 16379 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      warp10:
+        image: ghcr.io/${{ github.repository }}/warp10-ci:${{ github.sha }}
+        env:
+          standalone.port: '4802'
+          warpscript.maxops: '10000000'
+          ENABLE_SENSISION: 't'
+        options: >-
+          --health-cmd "curl localhost:4802/api/v0/check"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+          --health-start-period 60s
+        ports:
+        - 4802:4802
+        - 8082:8082
+        - 9718:9718
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        lfs: true
+    - uses: actions/setup-node@v4
+      with:
+        node-version: '16.13.2'
+        cache: yarn
+    - uses: actions/setup-python@v5
+      with:
+        python-version: '3.9'
+        cache: pip
+    - name: Install python deps
+      run: pip install -r requirements.txt
+    - name: install dependencies
+      run: yarn install --frozen-lockfile --network-concurrency 1
+    - name: ${{ matrix.test.name }}
+      run: ${{ matrix.test.command }}
+      env: ${{ matrix.test.env }}
+
+  tests-v2-with-vault:
+    needs:
+      - build-ci
+    runs-on: ubuntu-latest
+    env:
+      REINDEX_PYTHON_INTERPRETER: python3
+    services:
+      redis:
+        image: ghcr.io/${{ github.repository }}/redis-ci:${{ github.sha }}
+        ports:
+        - 6379:6379
+        - 9121:9121
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-replica:
+        image: ghcr.io/${{ github.repository }}/redis-replica-ci:${{ github.sha }}
+        ports:
+        - 6380:6380
+        options: >-
+          --health-cmd "redis-cli -p 6380 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-sentinel:
+        image: bitnami/redis-sentinel:7.2.4
+        env:
+          REDIS_MASTER_SET: scality-s3
+          REDIS_SENTINEL_PORT_NUMBER: '16379'
+          REDIS_SENTINEL_QUORUM: '1'
+        ports:
+        - 16379:16379
+        options: >-
+          --health-cmd "redis-cli -p 16379 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      warp10:
+        image: ghcr.io/${{ github.repository }}/warp10-ci:${{ github.sha }}
+        env:
+          standalone.port: '4802'
+          warpscript.maxops: '10000000'
+          ENABLE_SENSISION: 't'
+        ports:
+        - 4802:4802
+        - 8082:8082
+        - 9718:9718
+        options: >-
+          --health-cmd "curl localhost:4802/api/v0/check"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+          --health-start-period 60s
+      vault:
+        image: ghcr.io/${{ github.repository }}/vault-ci:${{ github.sha }}
+        ports:
+        - 8500:8500
+        - 8600:8600
+        - 8700:8700
+        - 8800:8800
+        options: >-
+          --health-cmd "curl http://localhost:8500/_/healthcheck"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        lfs: true
+    - uses: actions/setup-node@v4
+      with:
+        node-version: '16.13.2'
+        cache: yarn
+    - uses: actions/setup-python@v5
+      with:
+        python-version: '3.9'
+        cache: pip
+    - name: Install python deps
+      run: pip install -r requirements.txt
+    - name: install dependencies
+      run: yarn install --frozen-lockfile --network-concurrency 1
+    - name: Wait for warp10 for 60 seconds
+      run: sleep 60
+    - name: run v2 functional tests
+      run: bash ./.github/scripts/run_ft_tests.bash true ft_test:v2
+      env:
+        UTAPI_CACHE_BACKEND: redis
+        UTAPI_SERVICE_USER_ENABLED: 'true'
+        UTAPI_LOG_LEVEL: trace
+        SETUP_CMD: "run start_v2:server"
+    - name: 'Debug: SSH to runner'
+      uses: scality/actions/action-ssh-to-runner@1.7.0
+      timeout-minutes: ${{ fromJSON(github.event.inputs.connection-timeout-m) }}
+      continue-on-error: true
+      with:
+        tmate-server-host: ${{ secrets.TMATE_SERVER_HOST }}
+        tmate-server-port: ${{ secrets.TMATE_SERVER_PORT }}
+        tmate-server-rsa-fingerprint: ${{ secrets.TMATE_SERVER_RSA_FINGERPRINT }}
+        tmate-server-ed25519-fingerprint: ${{ secrets.TMATE_SERVER_ED25519_FINGERPRINT }}
+      if: ${{ ( github.event.inputs.debug == true || github.event.inputs.debug == 'true' ) }}
+
+  tests-v2-without-sensision:
+    needs:
+      - build-ci
+    runs-on: ubuntu-latest
+    env:
+      REINDEX_PYTHON_INTERPRETER: python3
+    name: ${{ matrix.test.name }}
+    strategy:
+      fail-fast: false
+      matrix:
+        test:
+        - name: run v2 soft limit test
+          command: bash ./.github/scripts/run_ft_tests.bash true ft_test:softLimit
+          env:
+            UTAPI_CACHE_BACKEND: redis
+            UTAPI_LOG_LEVEL: trace
+            SETUP_CMD: "run start_v2:server"
+        - name: run v2 hard limit test
+          command: bash ./.github/scripts/run_ft_tests.bash true ft_test:hardLimit
+          env:
+            UTAPI_CACHE_BACKEND: redis
+            UTAPI_LOG_LEVEL: trace
+            SETUP_CMD: "run start_v2:server"
+    services:
+      redis:
+        image: ghcr.io/${{ github.repository }}/redis-ci:${{ github.sha }}
+        ports:
+        - 6379:6379
+        - 9121:9121
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-replica:
+        image: ghcr.io/${{ github.repository }}/redis-replica-ci:${{ github.sha }}
+        ports:
+        - 6380:6380
+        options: >-
+          --health-cmd "redis-cli -p 6380 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      redis-sentinel:
+        image: bitnami/redis-sentinel:7.2.4
+        env:
+          REDIS_MASTER_SET: scality-s3
+          REDIS_SENTINEL_PORT_NUMBER: '16379'
+          REDIS_SENTINEL_QUORUM: '1'
+        ports:
+        - 16379:16379
+        options: >-
+          --health-cmd "redis-cli -p 16379 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      warp10:
+        image: ghcr.io/${{ github.repository }}/warp10-ci:${{ github.sha }}
+        env:
+          standalone.port: '4802'
+          warpscript.maxops: '10000000'
+        ports:
+        - 4802:4802
+        - 8082:8082
+        - 9718:9718
+        options: >-
+          --health-cmd "curl localhost:4802/api/v0/check"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+          --health-start-period 60s
+      vault:
+        image: ghcr.io/${{ github.repository }}/vault-ci:${{ github.sha }}
+        ports:
+        - 8500:8500
+        - 8600:8600
+        - 8700:8700
+        - 8800:8800
+        options: >-
+          --health-cmd "curl http://localhost:8500/_/healthcheck"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        lfs: true
+    - uses: actions/setup-node@v4
+      with:
+        node-version: '16.13.2'
+        cache: yarn
+    - uses: actions/setup-python@v5
+      with:
+        python-version: '3.9'
+        cache: pip
+    - name: Install python deps
+      run: pip install -r requirements.txt
+    - name: install dependencies
+      run: yarn install --frozen-lockfile --network-concurrency 1
+    - name: Wait for warp10 a little bit
+      run: sleep 60
+    - name: ${{ matrix.test.name }}
+      run: ${{ matrix.test.command }}
+      env: ${{ matrix.test.env }}
+    - name: 'Debug: SSH to runner'
+      uses: scality/actions/action-ssh-to-runner@1.7.0
+      timeout-minutes: ${{ fromJSON(github.event.inputs.connection-timeout-m) }}
+      continue-on-error: true
+      with:
+        tmate-server-host: ${{ secrets.TMATE_SERVER_HOST }}
+        tmate-server-port: ${{ secrets.TMATE_SERVER_PORT }}
+        tmate-server-rsa-fingerprint: ${{ secrets.TMATE_SERVER_RSA_FINGERPRINT }}
+        tmate-server-ed25519-fingerprint: ${{ secrets.TMATE_SERVER_ED25519_FINGERPRINT }}
+      if: ${{ ( github.event.inputs.debug == true || github.event.inputs.debug == 'true' ) }}

Dockerfile

@@ -0,0 +1,31 @@
+FROM node:16.13.2-buster-slim
+
+WORKDIR /usr/src/app
+
+COPY package.json yarn.lock /usr/src/app/
+
+RUN apt-get update \
+    && apt-get install -y \
+    curl \
+    gnupg2
+
+RUN curl -sS http://dl.yarnpkg.com/debian/pubkey.gpg | apt-key add - \
+    && echo "deb http://dl.yarnpkg.com/debian/ stable main" | tee /etc/apt/sources.list.d/yarn.list
+
+RUN apt-get update \
+    && apt-get install -y jq git python3 build-essential yarn --no-install-recommends \
+    && yarn cache clean \
+    && yarn install --frozen-lockfile --production --ignore-optional --network-concurrency=1 \
+    && apt-get autoremove --purge -y python3 git build-essential \
+    && rm -rf /var/lib/apt/lists/* \
+    && yarn cache clean \
+    && rm -rf ~/.node-gyp \
+    && rm -rf /tmp/yarn-*
+
+# Keep the .git directory in order to properly report version
+COPY . /usr/src/app
+
+ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]
+CMD [ "yarn", "start" ]
+
+EXPOSE 8100

README.md

@@ -3,9 +3,8 @@
 ![Utapi logo](res/utapi-logo.png)
 
 [![Circle CI][badgepub]](https://circleci.com/gh/scality/utapi)
-[![Scality CI][badgepriv]](http://ci.ironmann.io/gh/scality/utapi)
 
-Service Utilization API for tracking resource usage and metrics reporting
+Service Utilization API for tracking resource usage and metrics reporting.
 
 ## Design
 
@@ -88,13 +87,13 @@ Server is running.
 1. Create an IAM user
 
     ```
-    aws iam --endpoint-url <endpoint> create-user --user-name utapiuser
+    aws iam --endpoint-url <endpoint> create-user --user-name <user-name>
     ```
 
 2. Create access key for the user
 
     ```
-    aws iam --endpoint-url <endpoint> create-access-key --user-name utapiuser
+    aws iam --endpoint-url <endpoint> create-access-key --user-name <user-name>
     ```
 
 3. Define a managed IAM policy
@@ -203,12 +202,11 @@ Server is running.
 5. Attach user to the managed policy
 
     ```
-    aws --endpoint-url <endpoint> iam  attach-user-policy --user-name utapiuser
-    --policy-arn <policy arn>
+    aws --endpoint-url <endpoint> iam  attach-user-policy --user-name
+    <user-name> --policy-arn <policy arn>
     ```
 
-Now the user `utapiuser` has access to ListMetrics request in Utapi on all
-buckets.
+Now the user has access to ListMetrics request in Utapi on all buckets.
 
 ### Signing request with Auth V4
 
@@ -224,16 +222,18 @@ following urls for reference.
 You may also view examples making a request with Auth V4 using various languages
 and AWS SDKs [here](/examples).
 
-Alternatively, you can use a nifty command line tool available in Scality's S3.
+Alternatively, you can use a nifty command line tool available in Scality's
+CloudServer.
 
-You can git clone S3 repo from here https://github.com/scality/S3.git and follow
-the instructions in README to install the dependencies.
+You can git clone the CloudServer repo from here
+https://github.com/scality/cloudserver and follow the instructions in the README
+to install the dependencies.
 
-If you have S3 running inside a docker container you can docker exec into the S3
-container as
+If you have CloudServer running inside a docker container you can docker exec
+into the CloudServer container as
 
 ```
-docker exec -it <container id> bash
+docker exec -it <container-id> bash
 ```
 
 and then run the command
@@ -271,7 +271,7 @@ Usage: list_metrics [options]
     -v, --verbose
 ```
 
-A typical call to list metrics for a bucket `demo` to Utapi in a https enabled
+An example call to list metrics for a bucket `demo` to Utapi in a https enabled
 deployment would be
 
 ```
@@ -283,7 +283,7 @@ Both start and end times are time expressed as UNIX epoch timestamps **expressed
 in milliseconds**.
 
 Keep in mind, since Utapi metrics are normalized to the nearest 15 min.
-interval, so start time and end time need to be in specific format as follows.
+interval, start time and end time need to be in the specific format as follows.
 
 #### Start time
 
@@ -297,7 +297,7 @@ Date: Tue Oct 11 2016 17:35:25 GMT-0700 (PDT)
 
 Unix timestamp (milliseconds): 1476232525320
 
-Here's a typical JS method to get start timestamp
+Here's an example JS method to get a start timestamp
 
 ```javascript
 function getStartTimestamp(t) {
@@ -317,7 +317,7 @@ seconds and milliseconds set to 59 and 999 respectively. So valid end timestamps
 would look something like `09:14:59:999`, `09:29:59:999`, `09:44:59:999` and
 `09:59:59:999`.
 
-Here's a typical JS method to get end timestamp
+Here's an example JS method to get an end timestamp
 
 ```javascript
 function getEndTimestamp(t) {
@@ -342,4 +342,3 @@ In order to contribute, please follow the
 https://github.com/scality/Guidelines/blob/master/CONTRIBUTING.md).
 
 [badgepub]: http://circleci.com/gh/scality/utapi.svg?style=svg
-[badgepriv]: http://ci.ironmann.io/gh/scality/utapi.svg?style=svg

bin/createCheckpoint.js

@@ -1,12 +1,13 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'CreateCheckpoint',
 });
 
 
-const task = new tasks.CreateCheckpoint();
+const task = new tasks.CreateCheckpoint({ warp10: [warp10Clients[0]] });
 
 task.setup()
     .then(() => logger.info('Starting checkpoint creation'))

bin/createSnapshot.js

@@ -1,11 +1,12 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'CreateSnapshot',
 });
 
-const task = new tasks.CreateSnapshot();
+const task = new tasks.CreateSnapshot({ warp10: [warp10Clients[0]] });
 
 task.setup()
     .then(() => logger.info('Starting snapshot creation'))

bin/diskUsage.js

@@ -0,0 +1,15 @@
+const { tasks } = require('..');
+const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
+
+const logger = new LoggerContext({
+    task: 'MonitorDiskUsage',
+});
+
+
+const task = new tasks.MonitorDiskUsage({ warp10: [warp10Clients[0]] });
+
+task.setup()
+    .then(() => logger.info('Starting disk usage monitor'))
+    .then(() => task.start())
+    .then(() => logger.info('Disk usage monitor started'));

bin/downsample.js

@@ -1,14 +0,0 @@
-const { tasks } = require('..');
-const { LoggerContext } = require('../libV2/utils');
-
-const logger = new LoggerContext({
-    task: 'Downsample',
-});
-
-
-const task = new tasks.DownsampleTask();
-
-task.setup()
-    .then(() => logger.info('Starting Downsample daemon'))
-    .then(() => task.start())
-    .then(() => logger.info('Downsample started'));

bin/ensureServiceUser

@@ -0,0 +1,276 @@
+#! /usr/bin/env node
+
+// TODO
+// - deduplicate with Vault's seed script at https://github.com/scality/Vault/pull/1627
+// - add permission boundaries to user when https://scality.atlassian.net/browse/VAULT-4 is implemented
+
+const { errors } = require('arsenal');
+const program = require('commander');
+const werelogs = require('werelogs');
+const async = require('async');
+const { IAM } = require('aws-sdk');
+const { version } = require('../package.json');
+
+const systemPrefix = '/scality-internal/';
+
+function generateUserPolicyDocument() {
+    return {
+        Version: '2012-10-17',
+        Statement: {
+            Effect: 'Allow',
+            Action: 'utapi:ListMetrics',
+            Resource: 'arn:scality:utapi:::*/*',
+        },
+    };
+}
+
+function createIAMClient(opts) {
+    return new IAM({
+        endpoint: opts.iamEndpoint,
+    });
+}
+
+function needsCreation(v) {
+    if (Array.isArray(v)) {
+        return !v.length;
+    }
+
+    return !v;
+}
+
+class BaseHandler {
+    constructor(serviceName, iamClient, log) {
+        this.serviceName = serviceName;
+        this.iamClient = iamClient;
+        this.log = log;
+    }
+
+    applyWaterfall(values, done) {
+        this.log.debug('applyWaterfall', { values, type: this.resourceType });
+
+        const v = values[this.resourceType];
+
+        if (needsCreation(v)) {
+            this.log.debug('creating', { v, type: this.resourceType });
+            return this.create(values)
+                .then(res =>
+                    done(null, Object.assign(values, {
+                        [this.resourceType]: res,
+                    })))
+                .catch(done);
+        }
+
+        this.log.debug('conflicts check', { v, type: this.resourceType });
+        if (this.conflicts(v)) {
+            return done(errors.EntityAlreadyExists.customizeDescription(
+                `${this.resourceType} ${this.serviceName} already exists and conflicts with the expected value.`));
+        }
+
+        this.log.debug('nothing to do', { v, type: this.resourceType });
+        return done(null, values);
+    }
+}
+
+class UserHandler extends BaseHandler {
+    get resourceType() {
+        return 'user';
+    }
+
+    collect() {
+        return this.iamClient.getUser({
+            UserName: this.serviceName,
+        })
+        .promise()
+        .then(res => res.User);
+    }
+
+    create(allResources) {
+        return this.iamClient.createUser({
+            UserName: this.serviceName,
+            Path: systemPrefix,
+        })
+        .promise()
+        .then(res => res.User);
+    }
+
+    conflicts(u) {
+        return u.Path !== systemPrefix;
+    }
+}
+
+class PolicyHandler extends BaseHandler {
+    get resourceType() {
+        return 'policy';
+    }
+
+    collect() {
+        return this.iamClient.listPolicies({
+            MaxItems: 100,
+            OnlyAttached: false,
+            Scope: 'All',
+        })
+        .promise()
+        .then(res => res.Policies.find(p => p.PolicyName === this.serviceName));
+    }
+
+    create(allResources) {
+        const doc = generateUserPolicyDocument();
+
+        return this.iamClient.createPolicy({
+            PolicyName: this.serviceName,
+            PolicyDocument: JSON.stringify(doc),
+            Path: systemPrefix,
+        })
+        .promise()
+        .then(res => res.Policy);
+    }
+
+    conflicts(p) {
+        return p.Path !== systemPrefix;
+    }
+}
+
+class PolicyAttachmentHandler extends BaseHandler {
+    get resourceType() {
+        return 'policyAttachment';
+    }
+
+    collect() {
+        return this.iamClient.listAttachedUserPolicies({
+            UserName: this.serviceName,
+            MaxItems: 100,
+        })
+        .promise()
+        .then(res => res.AttachedPolicies)
+    }
+
+    create(allResources) {
+        return this.iamClient.attachUserPolicy({
+            PolicyArn: allResources.policy.Arn,
+            UserName: this.serviceName,
+        })
+        .promise();
+    }
+
+    conflicts(p) {
+        return false;
+    }
+}
+
+class AccessKeyHandler extends BaseHandler {
+    get resourceType() {
+        return 'accessKey';
+    }
+
+    collect() {
+        return this.iamClient.listAccessKeys({
+            UserName: this.serviceName,
+            MaxItems: 100,
+        })
+        .promise()
+        .then(res => res.AccessKeyMetadata)
+    }
+
+    create(allResources) {
+        return this.iamClient.createAccessKey({
+            UserName: this.serviceName,
+        })
+        .promise()
+        .then(res => res.AccessKey);
+    }
+
+    conflicts(a) {
+        return false;
+    }
+}
+
+function collectResource(v, done) {
+    v.collect()
+        .then(res => done(null, res))
+        .catch(err => {
+            if (err.code === 'NoSuchEntity') {
+                return done(null, null);
+            }
+
+            done(err);
+        });
+}
+
+function collectResourcesFromHandlers(handlers, cb) {
+    const tasks = handlers.reduce((acc, v) => ({
+        [v.resourceType]: done => collectResource(v, done),
+        ...acc,
+    }), {});
+    async.parallel(tasks, cb);
+}
+
+function buildServiceUserHandlers(serviceName, client, log) {
+    return [
+        UserHandler,
+        PolicyHandler,
+        PolicyAttachmentHandler,
+        AccessKeyHandler,
+    ].map(h => new h(serviceName, client, log));
+}
+
+function apply(client, serviceName, log, cb) {
+    const handlers = buildServiceUserHandlers(serviceName, client, log);
+
+    async.waterfall([
+        done => collectResourcesFromHandlers(handlers, done),
+        ...handlers.map(h => h.applyWaterfall.bind(h)),
+        (values, done) => done(null, values.accessKey),
+    ], cb);
+}
+
+function wrapAction(actionFunc, serviceName, options) {
+    werelogs.configure({
+        level: options.logLevel,
+        dump: options.logDumpLevel,
+    });
+
+    const log = new werelogs.Logger(process.argv[1]).newRequestLogger();
+    const client = createIAMClient(options);
+
+    actionFunc(client, serviceName, log, (err, data) => {
+        if (err) {
+            log.error('failed', {
+                data,
+                error: err,
+            });
+            if (err.EntityAlreadyExists) {
+                log.error(`run "${process.argv[1]} purge ${serviceName}" to fix.`);
+            }
+            process.exit(1);
+        }
+        log.info('success', { data });
+        process.exit();
+    });
+}
+
+program.version(version);
+
+[
+    {
+        name: 'apply <service-name>',
+        actionFunc: apply,
+    },
+].forEach(cmd => {
+    program
+        .command(cmd.name)
+        .option('--iam-endpoint <url>', 'IAM endpoint', 'http://localhost:8600')
+        .option('--log-level <level>', 'log level', 'info')
+        .option('--log-dump-level <level>', 'log level that triggers a dump of the debug buffer', 'error')
+        .action(wrapAction.bind(null, cmd.actionFunc));
+});
+
+const validCommands = program.commands.map(n => n._name);
+
+// Is the command given invalid or are there too few arguments passed
+if (!validCommands.includes(process.argv[2])) {
+    program.outputHelp();
+    process.stdout.write('\n');
+    process.exit(1);
+} else {
+    program.parse(process.argv);
+}

bin/ingestShards.js

@@ -1,12 +1,13 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'IngestShard',
 });
 
 
-const task = new tasks.IngestShard();
+const task = new tasks.IngestShard({ warp10: warp10Clients });
 
 task.setup()
     .then(() => logger.info('Starting shard ingestion'))

bin/manualAdjust.js

@@ -0,0 +1,15 @@
+const { tasks } = require('..');
+const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
+
+const logger = new LoggerContext({
+    task: 'ManualAdjust',
+});
+
+
+const task = new tasks.ManualAdjust({ warp10: warp10Clients });
+
+task.setup()
+    .then(() => logger.info('Starting manual adjustment'))
+    .then(() => task.start())
+    .then(() => logger.info('Manual adjustment started'));

bin/migrate.js

@@ -1,11 +1,12 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'Migrate',
 });
 
-const task = new tasks.MigrateTask();
+const task = new tasks.MigrateTask({ warp10: [warp10Clients[0]] });
 
 task.setup()
     .then(() => logger.info('Starting utapi v1 => v2 migration'))

bin/reindex.js

@@ -1,12 +1,13 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'Reindex',
 });
 
 
-const task = new tasks.ReindexTask();
+const task = new tasks.ReindexTask({ warp10: [warp10Clients[0]] });
 
 task.setup()
     .then(() => logger.info('Starting Reindex daemon'))

bin/repair.js

@@ -1,12 +1,13 @@
 const { tasks } = require('..');
 const { LoggerContext } = require('../libV2/utils');
+const { clients: warp10Clients } = require('../libV2/warp10');
 
 const logger = new LoggerContext({
     task: 'Repair',
 });
 
 
-const task = new tasks.RepairTask();
+const task = new tasks.RepairTask({ warp10: [warp10Clients[0]] });
 
 task.setup()
     .then(() => logger.info('Starting Repair daemon'))

docker-compose.yaml

@@ -0,0 +1,75 @@
+version: '3.8'
+
+x-models:
+  warp10: &warp10
+    build:
+      context: .
+      dockerfile: ./images/warp10/Dockerfile
+    volumes: [ $PWD/warpscript:/usr/local/share/warpscript ]
+
+  warp10_env: &warp10_env
+    ENABLE_WARPSTUDIO: 'true'
+    ENABLE_SENSISION: 'true'
+    warpscript.repository.refresh: 1000
+    warpscript.maxops: 1000000000
+    warpscript.maxops.hard: 1000000000
+    warpscript.maxfetch: 1000000000
+    warpscript.maxfetch.hard: 1000000000
+    warpscript.extension.debug: io.warp10.script.ext.debug.DebugWarpScriptExtension
+    warpscript.maxrecursion: 1000
+    warpscript.repository.directory: /usr/local/share/warpscript
+    warpscript.extension.logEvent: io.warp10.script.ext.logging.LoggingWarpScriptExtension
+
+  redis: &redis
+      build:
+        context: .
+        dockerfile: ./images/redis/Dockerfile
+
+services:
+  redis-0:
+    image: redis:7.2.4
+    command: redis-server --port 6379 --slave-announce-ip "${EXTERNAL_HOST}"
+    ports:
+      - 6379:6379
+    environment:
+      - HOST_IP="${EXTERNAL_HOST}"
+
+  redis-1:
+    image: redis:7.2.4
+    command: redis-server --port 6380 --slaveof "${EXTERNAL_HOST}" 6379 --slave-announce-ip "${EXTERNAL_HOST}"
+    ports:
+       - 6380:6380
+    environment:
+      - HOST_IP="${EXTERNAL_HOST}"
+
+  redis-sentinel-0:
+    image: redis:7.2.4
+    command: |-
+      bash -c 'cat > /tmp/sentinel.conf <<EOF
+      port 16379
+      logfile ""
+      dir /tmp
+      sentinel announce-ip ${EXTERNAL_HOST}
+      sentinel announce-port 16379
+      sentinel monitor scality-s3 "${EXTERNAL_HOST}" 6379 1
+      EOF
+      redis-sentinel /tmp/sentinel.conf'
+
+    environment:
+      - HOST_IP="${EXTERNAL_HOST}"
+    ports:
+      - 16379:16379
+
+  warp10:
+    << : *warp10
+    environment:
+      << : *warp10_env
+
+    ports:
+      - 4802:4802
+      - 8081:8081
+      - 9718:9718
+
+    volumes:
+      - /tmp/warp10:/data
+      - '${PWD}/warpscript:/usr/local/share/warpscript'

docker-entrypoint.sh

@@ -0,0 +1,47 @@
+#!/bin/bash
+
+# set -e stops the execution of a script if a command or pipeline has an error
+set -e
+
+# modifying config.json
+JQ_FILTERS_CONFIG="."
+
+if [[ "$LOG_LEVEL" ]]; then
+    if [[ "$LOG_LEVEL" == "info" || "$LOG_LEVEL" == "debug" || "$LOG_LEVEL" == "trace" ]]; then
+        JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .log.logLevel=\"$LOG_LEVEL\""
+        echo "Log level has been modified to $LOG_LEVEL"
+    else
+        echo "The log level you provided is incorrect (info/debug/trace)"
+    fi
+fi
+
+if [[ "$WORKERS" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .workers=\"$WORKERS\""
+fi
+
+if [[ "$REDIS_HOST" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .redis.host=\"$REDIS_HOST\""
+fi
+
+if [[ "$REDIS_PORT" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .redis.port=\"$REDIS_PORT\""
+fi
+
+if [[ "$VAULTD_HOST" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .vaultd.host=\"$VAULTD_HOST\""
+fi
+
+if [[ "$VAULTD_PORT" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .vaultd.port=\"$VAULTD_PORT\""
+fi
+
+if [[ "$HEALTHCHECKS_ALLOWFROM" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .healthChecks.allowFrom=[\"$HEALTHCHECKS_ALLOWFROM\"]"
+fi
+
+if [[ $JQ_FILTERS_CONFIG != "." ]]; then
+    jq "$JQ_FILTERS_CONFIG" config.json > config.json.tmp
+    mv config.json.tmp config.json
+fi
+
+exec "$@"

docs/RELEASE.md

@@ -0,0 +1,42 @@
+# Utapi Release Plan
+
+## Docker Image Generation
+
+Docker images are hosted on [ghcr.io](https://github.com/orgs/scality/packages).
+Utapi has one namespace there:
+
+* Namespace: ghcr.io/scality/utapi
+
+With every CI build, the CI will push images, tagging the
+content with the developer branch's short SHA-1 commit hash.
+This allows those images to be used by developers, CI builds,
+build chain and so on.
+
+Tagged versions of utapi will be stored in the production namespace.
+
+## How to Pull Docker Images
+
+```sh
+docker pull ghcr.io/scality/utapi:<commit hash>
+docker pull ghcr.io/scality/utapi:<tag>
+```
+
+## Release Process
+
+To release a production image:
+
+* Name the tag for the repository and Docker image.
+
+* Use the `yarn version` command with the same tag to update `package.json`.
+
+* Create a PR and merge the `package.json` change.
+
+* Tag the repository using the same tag.
+
+* [Force a build] using:
+  * A given branch that ideally matches the tag.
+  * The `release` stage.
+  * An extra property with the name `tag` and its value being the actual tag.
+
+[Force a build]:
+https://eve.devsca.com/github/scality/utapi/#/builders/bootstrap/force/force

eve/main.yml

@@ -1,141 +0,0 @@
----
-version: 0.2
-
-branches:
-  default:
-    stage: pre-merge
-
-models:
-  - Git: &clone
-      name: Pull repo
-      repourl: '%(prop:git_reference)s'
-      shallow: True
-      retryFetch: True
-      haltOnFailure: True
-  - Workspace: &workspace
-      type: kube_pod
-      path: eve/workers/pod.yml
-      images:
-        aggressor:
-          context: '.'
-          dockerfile: eve/workers/unit_and_feature_tests/Dockerfile
-        warp10:
-          context: '.'
-          dockerfile: 'images/warp10/Dockerfile'
-        vault: eve/workers/mocks/vault
-  - Upload: &upload_artifacts
-      source: /artifacts
-      urls:
-        - "*"
-
-
-stages:
-  pre-merge:
-    worker:
-      type: local
-    steps:
-    - MasterShellCommand:
-        name: Replace upstream image with `-ci` variant
-        command: "sed -i '/^FROM/ s/$/-ci/' %(prop:master_builddir)s/build/images/warp10/Dockerfile"
-    - TriggerStages:
-        name: trigger all the tests
-        stage_names:
-        - linting-coverage
-        - run-unit-tests
-        - run-client-tests
-        - run-server-tests
-        - run-cron-tests
-        - run-interval-tests
-        - run-v2-functional-tests
-  linting-coverage:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run static analysis tools on markdown
-          command: yarn run lint_md
-      - ShellCommand:
-          name: run static analysis tools on code
-          command: yarn run lint
-  run-unit-tests:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run unit tests
-          command: yarn test
-  run-client-tests:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run client tests
-          command: bash ./eve/workers/unit_and_feature_tests/run_ft_tests.bash false ft_test:client
-          logfiles:
-            utapi:
-              filename: "/artifacts/setup_ft_test:client.log"
-              follow: true
-  run-server-tests:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run server tests
-          command: bash ./eve/workers/unit_and_feature_tests/run_ft_tests.bash false ft_test:server
-          logfiles:
-            utapi:
-              filename: "/artifacts/setup_ft_test:server.log"
-              follow: true
-  run-cron-tests:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run cron tests
-          command: bash ./eve/workers/unit_and_feature_tests/run_ft_tests.bash false ft_test:cron
-          logfiles:
-            utapi:
-              filename: "/artifacts/setup_ft_test:cron.log"
-              follow: true
-  run-interval-tests:
-    worker: *workspace
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: run interval tests
-          command: bash ./eve/workers/unit_and_feature_tests/run_ft_tests.bash true ft_test:interval
-          logfiles:
-            utapi:
-              filename: "/artifacts/setup_ft_test:interval.log"
-              follow: true
-  run-v2-functional-tests:
-    worker:
-      << : *workspace
-      vars:
-        vault: enabled
-    steps:
-      - Git: *clone
-      - ShellCommand:
-          name: Wait for Warp 10
-          command: |
-            bash -c "
-            set -ex
-            bash tests/utils/wait_for_local_port.bash 4802 60"
-          logfiles:
-            warp10:
-              filename: "/artifacts/warp10.log"
-              follow: true
-      - ShellCommand:
-          name: run v2 functional tests
-          command: SETUP_CMD="run start_v2:server" bash ./eve/workers/unit_and_feature_tests/run_ft_tests.bash true ft_test:v2
-          env:
-            UTAPI_CACHE_BACKEND: redis
-            UTAPI_LOG_LEVEL: trace
-          logfiles:
-            warp10:
-              filename: "/artifacts/warp10.log"
-              follow: true
-            utapi:
-              filename: "/artifacts/setup_ft_test:v2.log"
-              follow: true
-      - Upload: *upload_artifacts

eve/workers/mocks/vault/Dockerfile

@@ -1,7 +0,0 @@
-FROM node:alpine
-
-ADD ./vault.js /usr/share/src/
-
-WORKDIR /usr/share/src/
-
-CMD node vault.js

eve/workers/mocks/vault/vault.js

@@ -1,32 +0,0 @@
-const http = require('http');
-
-const port = process.env.VAULT_PORT || 8500;
-
-class Vault {
-    constructor() {
-        this._server = null;
-    }
-
-    static _onRequest(req, res) {
-        res.writeHead(200);
-        return res.end();
-    }
-
-    start() {
-        this._server = http.createServer(Vault._onRequest).listen(port);
-    }
-
-    end() {
-        this._server.close();
-    }
-}
-
-const vault = new Vault();
-
-['SIGINT', 'SIGQUIT', 'SIGTERM'].forEach(eventName => {
-    process.on(eventName, () => process.exit(0));
-});
-
-// eslint-disable-next-line no-console
-console.log('Starting Vault Mock...');
-vault.start();

eve/workers/pod.yml

@@ -1,67 +0,0 @@
-apiVersion: v1
-kind: Pod
-metadata:
-  name: "utapi-test-pod"
-spec:
-  activeDeadlineSeconds: 3600
-  restartPolicy: Never
-  terminationGracePeriodSeconds: 10
-  containers:
-  - name: aggressor
-    image: "{{ images.aggressor }}"
-    imagePullPolicy: IfNotPresent
-    resources:
-      requests:
-        cpu: 500m
-        memory: 1Gi
-      limits:
-        cpu: "2"
-        memory: 3Gi
-    volumeMounts:
-    - mountPath: /var/run/docker.sock
-      name: docker-socket
-    - name: artifacts
-      readOnly: false
-      mountPath: /artifacts
-  - name: warp10
-    image: "{{ images.warp10 }}"
-    command:
-      - sh
-      - -ce
-      - /init | tee -a /artifacts/warp10.log
-    env:
-      - name: standalone.port
-        value: '4802'
-      - name: warp.token.file
-        value: /opt/warp10/etc/ci.tokens
-      - name: warpscript.maxops
-        value: '10000000'
-    resources:
-      requests:
-        cpu: 500m
-        memory: 1Gi
-      limits:
-        cpu: 1750m
-        memory: 3Gi
-    volumeMounts:
-    - name: artifacts
-      readOnly: false
-      mountPath: /artifacts
-{% if vars.vault is defined and vars.vault == 'enabled' %}
-  - name: vault
-    image: "{{ images.vault }}"
-    resources:
-      requests:
-        cpu: 10m
-        memory: 64Mi
-      limits:
-        cpu: 50m
-        memory: 128Mi
-{% endif %}
-  volumes:
-  - name: docker-socket
-    hostPath:
-      path: /var/run/docker.sock
-      type: Socket
-  - name: artifacts
-    emptyDir: {}

eve/workers/unit_and_feature_tests/Dockerfile

@@ -1,48 +0,0 @@
-FROM buildpack-deps:jessie-curl
-
-#
-# Install apt packages needed by utapi and buildbot_worker
-#
-
-ENV LANG C.UTF-8
-ENV NODE_VERSION 10.22.0
-ENV PATH=$PATH:/utapi/node_modules/.bin
-ENV NODE_PATH=/utapi/node_modules
-
-COPY eve/workers/unit_and_feature_tests/utapi_packages.list eve/workers/unit_and_feature_tests/buildbot_worker_packages.list /tmp/
-
-WORKDIR /utapi
-
-RUN wget https://nodejs.org/dist/v${NODE_VERSION}/node-v${NODE_VERSION}-linux-x64.tar.gz \
-    && tar -xf node-v${NODE_VERSION}-linux-x64.tar.gz --directory /usr/local --strip-components 1 \
-    && curl -sS http://dl.yarnpkg.com/debian/pubkey.gpg | apt-key add - \
-    && echo "deb http://dl.yarnpkg.com/debian/ stable main" | tee /etc/apt/sources.list.d/yarn.list \
-    && apt-get update -qq \
-    && cat /tmp/*packages.list | xargs apt-get install -y \
-    && pip install pip==9.0.1 \
-    && rm -rf /var/lib/apt/lists/* \
-    && rm -f /tmp/*packages.list \
-    && rm -f /etc/supervisor/conf.d/*.conf \
-    && rm -f node-v${NODE_VERSION}-linux-x64.tar.gz
-
-#
-# Install yarn dependencies
-#
-
-COPY package.json yarn.lock /utapi/
-
-RUN yarn cache clean \
-    && yarn install --frozen-lockfile \
-    && yarn cache clean
-#
-# Run buildbot-worker on startup through supervisor
-#
-ARG BUILDBOT_VERSION
-
-RUN pip install buildbot-worker==$BUILDBOT_VERSION
-RUN pip3 install requests
-RUN pip3 install redis
-ADD eve/workers/unit_and_feature_tests/supervisor/buildbot_worker.conf /etc/supervisor/conf.d/
-ADD eve/workers/unit_and_feature_tests/redis/sentinel.conf /etc/sentinel.conf
-
-CMD ["supervisord", "-n"]

eve/workers/unit_and_feature_tests/buildbot_worker_packages.list

@@ -1,11 +0,0 @@
-ca-certificates
-git
-libffi-dev
-libssl-dev
-python2.7
-python2.7-dev
-python-pip
-sudo
-supervisor
-lsof
-netcat

eve/workers/unit_and_feature_tests/redis/sentinel.conf

@@ -1,35 +0,0 @@
-# Example sentinel.conf
-
-# The port that this sentinel instance will run on
-port 16379
-
-# Specify the log file name. Also the empty string can be used to force
-# Sentinel to log on the standard output. Note that if you use standard
-# output for logging but daemonize, logs will be sent to /dev/null
-logfile ""
-
-# dir <working-directory>
-# Every long running process should have a well-defined working directory.
-# For Redis Sentinel to chdir to /tmp at startup is the simplest thing
-# for the process to don't interfere with administrative tasks such as
-# unmounting filesystems.
-dir /tmp
-
-# sentinel monitor <master-name> <ip> <redis-port> <quorum>
-#
-# Tells Sentinel to monitor this master, and to consider it in O_DOWN
-# (Objectively Down) state only if at least <quorum> sentinels agree.
-#
-# Note that whatever is the ODOWN quorum, a Sentinel will require to
-# be elected by the majority of the known Sentinels in order to
-# start a failover, so no failover can be performed in minority.
-#
-# Replicas are auto-discovered, so you don't need to specify replicas in
-# any way. Sentinel itself will rewrite this configuration file adding
-# the replicas using additional configuration options.
-# Also note that the configuration file is rewritten when a
-# replica is promoted to master.
-#
-# Note: master name should not include special characters or spaces.
-# The valid charset is A-z 0-9 and the three characters ".-_".
-sentinel monitor scality-s3 127.0.0.1 6379 1

eve/workers/unit_and_feature_tests/supervisor/buildbot_worker.conf

@@ -1,14 +0,0 @@
-[program:buildbot_worker]
-command=/bin/sh -c 'buildbot-worker create-worker . "%(ENV_BUILDMASTER)s:%(ENV_BUILDMASTER_PORT)s" "%(ENV_WORKERNAME)s" "%(ENV_WORKERPASS)s"  && buildbot-worker start --nodaemon'
-autostart=true
-autorestart=false
-
-[program:redis_server]
-command=/usr/bin/redis-server
-autostart=true
-autorestart=false
-
-[program:redis_sentinel]
-command=/usr/bin/redis-server /etc/sentinel.conf --sentinel
-autostart=true
-autorestart=false

eve/workers/unit_and_feature_tests/utapi_packages.list

@@ -1,5 +0,0 @@
-build-essential
-redis-server
-python3
-python3-pip
-yarn

examples/python-V4Auth-example.py

@@ -0,0 +1,90 @@
+import sys, os, base64, datetime, hashlib, hmac, datetime, calendar, json
+import requests # pip install requests
+
+access_key = '9EQTVVVCLSSG6QBMNKO5'
+secret_key = 'T5mK/skkkwJ/mTjXZnHyZ5UzgGIN=k9nl4dyTmDH'
+
+method = 'POST'
+service = 's3'
+host = 'localhost:8100'
+region = 'us-east-1'
+canonical_uri = '/buckets'
+canonical_querystring = 'Action=ListMetrics&Version=20160815'
+content_type = 'application/x-amz-json-1.0'
+algorithm = 'AWS4-HMAC-SHA256'
+
+t = datetime.datetime.utcnow()
+amz_date = t.strftime('%Y%m%dT%H%M%SZ')
+date_stamp = t.strftime('%Y%m%d')
+
+# Key derivation functions. See:
+# http://docs.aws.amazon.com/general/latest/gr/signature-v4-examples.html#signature-v4-examples-python
+def sign(key, msg):
+    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
+
+def getSignatureKey(key, date_stamp, regionName, serviceName):
+    kDate = sign(('AWS4' + key).encode('utf-8'), date_stamp)
+    kRegion = sign(kDate, regionName)
+    kService = sign(kRegion, serviceName)
+    kSigning = sign(kService, 'aws4_request')
+    return kSigning
+
+def get_start_time(t):
+    start = t.replace(minute=t.minute - t.minute % 15, second=0, microsecond=0)
+    return calendar.timegm(start.utctimetuple()) * 1000;
+
+def get_end_time(t):
+    end = t.replace(minute=t.minute - t.minute % 15, second=0, microsecond=0)
+    return calendar.timegm(end.utctimetuple()) * 1000 - 1;
+
+start_time = get_start_time(datetime.datetime(2016, 1, 1, 0, 0, 0, 0))
+end_time = get_end_time(datetime.datetime(2016, 2, 1, 0, 0, 0, 0))
+
+# Request parameters for listing Utapi bucket metrics--passed in a JSON block.
+bucketListing = {
+    'buckets': [ 'utapi-test' ],
+    'timeRange': [ start_time, end_time ],
+}
+
+request_parameters = json.dumps(bucketListing)
+
+payload_hash = hashlib.sha256(request_parameters).hexdigest()
+
+canonical_headers = \
+    'content-type:{0}\nhost:{1}\nx-amz-content-sha256:{2}\nx-amz-date:{3}\n' \
+    .format(content_type, host, payload_hash, amz_date)
+
+signed_headers = 'content-type;host;x-amz-content-sha256;x-amz-date'
+
+canonical_request = '{0}\n{1}\n{2}\n{3}\n{4}\n{5}' \
+    .format(method, canonical_uri, canonical_querystring, canonical_headers,
+            signed_headers, payload_hash)
+
+credential_scope = '{0}/{1}/{2}/aws4_request' \
+    .format(date_stamp, region, service)
+
+string_to_sign = '{0}\n{1}\n{2}\n{3}' \
+    .format(algorithm, amz_date, credential_scope,
+            hashlib.sha256(canonical_request).hexdigest())
+
+signing_key = getSignatureKey(secret_key, date_stamp, region, service)
+
+signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'),
+                     hashlib.sha256).hexdigest()
+
+authorization_header = \
+    '{0} Credential={1}/{2}, SignedHeaders={3}, Signature={4}' \
+    .format(algorithm, access_key, credential_scope, signed_headers, signature)
+
+# The 'host' header is added automatically by the Python 'requests' library.
+headers = {
+    'Content-Type': content_type,
+    'X-Amz-Content-Sha256': payload_hash,
+    'X-Amz-Date': amz_date,
+    'Authorization': authorization_header
+}
+
+endpoint = 'http://' + host + canonical_uri + '?' + canonical_querystring;
+
+r = requests.post(endpoint, data=request_parameters, headers=headers)
+print (r.text)

images/nodesvc-base/Dockerfile

@@ -0,0 +1,20 @@
+FROM ghcr.io/scality/federation/nodesvc-base:7.10.5.0
+
+ENV UTAPI_CONFIG_FILE=${CONF_DIR}/config.json
+
+WORKDIR ${HOME_DIR}/utapi
+
+COPY ./package.json ./yarn.lock ${HOME_DIR}/utapi
+
+# Remove when gitcache is sorted out
+RUN rm /root/.gitconfig
+
+RUN yarn install --production --frozen-lockfile --network-concurrency 1
+
+COPY . ${HOME_DIR}/utapi
+
+RUN chown -R ${USER} ${HOME_DIR}/utapi
+
+USER ${USER}
+
+CMD bash -c "source ${CONF_DIR}/env && export && supervisord -c ${CONF_DIR}/${SUPERVISORD_CONF}"

images/redis/Dockerfile

@@ -0,0 +1,17 @@
+FROM redis:alpine
+
+ENV S6_VERSION 2.0.0.1
+ENV EXPORTER_VERSION 1.24.0
+ENV S6_BEHAVIOUR_IF_STAGE2_FAILS 2
+
+RUN wget https://github.com/just-containers/s6-overlay/releases/download/v${S6_VERSION}/s6-overlay-amd64.tar.gz -O /tmp/s6-overlay-amd64.tar.gz \
+    && tar xzf /tmp/s6-overlay-amd64.tar.gz -C / \
+    && rm -rf /tmp/s6-overlay-amd64.tar.gz 
+
+RUN wget https://github.com/oliver006/redis_exporter/releases/download/v${EXPORTER_VERSION}/redis_exporter-v${EXPORTER_VERSION}.linux-amd64.tar.gz -O redis_exporter.tar.gz \
+    && tar xzf redis_exporter.tar.gz -C / \
+    && cd .. \
+    && mv /redis_exporter-v${EXPORTER_VERSION}.linux-amd64/redis_exporter /usr/local/bin/redis_exporter
+
+ADD ./images/redis/s6 /etc
+CMD /init

images/redis/s6/services.d/exporter/run

@@ -0,0 +1,4 @@
+#!/usr/bin/with-contenv sh
+echo "starting redis exporter"
+exec redis_exporter
+

images/redis/s6/services.d/redis/run

@@ -0,0 +1,4 @@
+#!/usr/bin/with-contenv sh
+echo "starting redis"
+exec redis-server
+

images/warp10/90-default-host-port.conf

@@ -0,0 +1,2 @@
+standalone.host = 0.0.0.0
+standalone.port = 4802

images/warp10/Dockerfile

@@ -1,29 +1,56 @@
-FROM warp10io/warp10:2.6.0
+FROM golang:1.14-alpine as builder
+
+ENV WARP10_EXPORTER_VERSION 2.7.5
+
+RUN apk add zip unzip build-base \
+    && wget -q -O exporter.zip https://github.com/centreon/warp10-sensision-exporter/archive/refs/heads/master.zip \
+    && unzip exporter.zip \
+    && cd warp10-sensision-exporter-master \
+    && go mod download \
+    && cd tools \
+    && go run generate_sensision_metrics.go ${WARP10_EXPORTER_VERSION} \
+    && cp sensision.go ../collector/ \
+    && cd .. \
+    && go build -a -o /usr/local/go/warp10_sensision_exporter
+
+FROM ghcr.io/scality/utapi/warp10:2.8.1-95-g73e7de80
+
+# Override baked in version
+# Remove when updating to a numbered release
+ENV WARP10_VERSION 2.8.1-95-g73e7de80
 
 ENV S6_VERSION 2.0.0.1
+ENV S6_BEHAVIOUR_IF_STAGE2_FAILS 2
 
 ENV WARP10_CONF_TEMPLATES ${WARP10_HOME}/conf.templates/standalone
 ENV SENSISION_DATA_DIR /data/sensision
+ENV SENSISION_PORT 8082
 
 # Modify Warp 10 default config
-ENV standalone.host 0.0.0.0
-ENV standalone.port 4802
+ENV standalone.home /opt/warp10
 ENV warpscript.repository.directory /usr/local/share/warpscript
 ENV warp.token.file /static.tokens
 ENV warpscript.extension.protobuf io.warp10.ext.protobuf.ProtobufWarpScriptExtension
 ENV warpscript.extension.macrovalueencoder 'io.warp10.continuum.ingress.MacroValueEncoder$Extension'
-ENV warpscript.extension.concurrent 'io.warp10.script.ext.concurrent.ConcurrentWarpScriptExtension'
-
 # ENV warpscript.extension.debug io.warp10.script.ext.debug.DebugWarpScriptExtension
 
 RUN wget https://github.com/just-containers/s6-overlay/releases/download/v${S6_VERSION}/s6-overlay-amd64.tar.gz -O /tmp/s6-overlay-amd64.tar.gz \
     && tar xzf /tmp/s6-overlay-amd64.tar.gz -C / \
     && rm -rf /tmp/s6-overlay-amd64.tar.gz
 
+# Install jmx exporter
+ADD https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.16.1/jmx_prometheus_javaagent-0.16.1.jar /opt/jmx_prom_agent.jar
+ADD ./images/warp10/jmx_prom.yaml /opt/jmx_prom.yaml
+
 # Install protobuf extestion
-ADD https://dl.bintray.com/senx/maven/io/warp10/warp10-ext-protobuf/1.1.0-uberjar/warp10-ext-protobuf-1.1.0-uberjar.jar /opt/warp10/lib/
+ADD ./images/warp10/warp10-ext-protobuf-1.2.2-uberjar.jar /opt/warp10/lib/
+
+# Install Sensision exporter
+COPY --from=builder /usr/local/go/warp10_sensision_exporter /usr/local/bin/warp10_sensision_exporter
 
 ADD ./images/warp10/s6 /etc
 ADD ./warpscript /usr/local/share/warpscript
 ADD ./images/warp10/static.tokens /
+ADD ./images/warp10/90-default-host-port.conf $WARP10_CONF_TEMPLATES/90-default-host-port.conf
+
 CMD /init

images/warp10/jmx_prom.yaml

@@ -0,0 +1,2 @@
+---
+startDelaySeconds: 0

images/warp10/s6/cont-init.d/10-generate-hash-seeds

@@ -2,10 +2,13 @@
 
 set -eu
 
+WARP10_JAR=${WARP10_HOME}/bin/warp10-${WARP10_VERSION}.jar
 WARP10_CONFIG_DIR="$WARP10_DATA_DIR/conf"
 WARP10_SECRETS="$WARP10_CONFIG_DIR/00-secrets.conf"
 
 if [ ! -f "$WARP10_SECRETS" ]; then
     cp "$WARP10_CONF_TEMPLATES/00-secrets.conf.template" "$WARP10_SECRETS"
-    python "${WARP10_HOME}/etc/generate_crypto_key.py" "$WARP10_SECRETS"
+    /usr/bin/java -cp ${WARP10_JAR} -Dfile.encoding=UTF-8 io.warp10.GenerateCryptoKey ${WARP10_SECRETS}
+    echo "warp10.manager.secret = scality" >> $WARP10_SECRETS
 fi
+

images/warp10/s6/cont-init.d/40-generate-secrets

@@ -1,28 +0,0 @@
-#!/usr/bin/with-contenv sh
-
-set -eu
-
-JAVA="/usr/bin/java"
-WARP10_JAR=${WARP10_HOME}/bin/warp10-${WARP10_VERSION}.jar
-WARP10_CP="${WARP10_HOME}/etc:${WARP10_JAR}:${WARP10_HOME}/lib/*"
-
-WARP10_CONFIG_DIR="$WARP10_DATA_DIR/conf"
-INITIAL_TOKENS="$WARP10_CONFIG_DIR/initial.token"
-
-if [ ! -f "$INITIAL_TOKENS" ]; then
-    CONFIG_FILES="$(find ${WARP10_CONFIG_DIR} -not -path "*/\.*" -name "*.conf" | sort | tr '\n' ' ' 2> /dev/null)"
-
-    # Look for a set token secret and use it for generation
-    secret=`${JAVA} -cp ${WARP10_CP} io.warp10.WarpConfig ${CONFIG_FILES} . 'token.secret' | sed -n 's/^@CONF@ //p' | sed -n 's/^token.secret[^=]*=//p'`
-    if [[ "${secret}"  != "null" ]]; then
-        sed -i.bak -e "s|^{{secret}}|'"${secret}"'|" ${WARP10_HOME}/templates/warp10-tokengen.mc2
-    else
-        sed -i.bak -e "s|^{{secret}}||" ${WARP10_HOME}/templates/warp10-tokengen.mc2
-    fi
-    rm ${WARP10_HOME}/templates/warp10-tokengen.mc2.bak
-
-    # Generate read/write tokens valid for a period of 100 years. We use 'io.warp10.bootstrap' as application name.
-    ${JAVA} -cp ${WARP10_JAR} io.warp10.worf.TokenGen ${CONFIG_FILES} ${WARP10_HOME}/templates/warp10-tokengen.mc2 $INITIAL_TOKENS
-    sed -i.bak 's/^.\{1\}//;$ s/.$//' $INITIAL_TOKENS # Remove first and last character
-    rm "${INITIAL_TOKENS}.bak"
-fi

images/warp10/s6/services.d/sensision-exporter/run

@@ -0,0 +1,12 @@
+#!/usr/bin/with-contenv sh
+
+EXPORTER_CMD="warp10_sensision_exporter --warp10.url=http://localhost:${SENSISION_PORT}/metrics"
+
+if [ -f "/usr/local/bin/warp10_sensision_exporter" -a -n "$ENABLE_SENSISION" ]; then
+    echo "Starting Sensision exporter with $EXPORTER_CMD ..."
+    exec $EXPORTER_CMD
+else
+    echo "Sensision is disabled. Not starting exporter."
+    # wait indefinitely
+    exec tail -f /dev/null
+fi

images/warp10/s6/services.d/sensision/run

@@ -3,9 +3,8 @@
 JAVA="/usr/bin/java"
 JAVA_OPTS=""
 
-VERSION=1.0.21
 SENSISION_CONFIG=${SENSISION_DATA_DIR}/conf/sensision.conf
-SENSISION_JAR=${SENSISION_HOME}/bin/sensision-${VERSION}.jar
+SENSISION_JAR=${SENSISION_HOME}/bin/sensision-${SENSISION_VERSION}.jar
 SENSISION_CP=${SENSISION_HOME}/etc:${SENSISION_JAR}
 SENSISION_CLASS=io.warp10.sensision.Main
 export MALLOC_ARENA_MAX=1
@@ -14,13 +13,13 @@ if [ -z "$SENSISION_HEAP" ]; then
     SENSISION_HEAP=64m
 fi
 
-SENSISION_CMD="${JAVA} ${JAVA_OPTS} -Xmx${SENSISION_HEAP} -Dsensision.server.port=0 ${SENSISION_OPTS} -Dsensision.config=${SENSISION_CONFIG} -cp ${SENSISION_CP} ${SENSISION_CLASS}"
+SENSISION_CMD="${JAVA} ${JAVA_OPTS} -Xmx${SENSISION_HEAP} -Dsensision.server.port=${SENSISION_PORT} ${SENSISION_OPTS} -Dsensision.config=${SENSISION_CONFIG} -cp ${SENSISION_CP} ${SENSISION_CLASS}"
 
 if [ -n "$ENABLE_SENSISION" ]; then
     echo "Starting Sensision with $SENSISION_CMD ..."
     exec $SENSISION_CMD | tee -a ${SENSISION_HOME}/logs/sensision.log
 else
-    echo "Sensision is disabled"
+    echo "Sensision is disabled. Not starting."
     # wait indefinitely
     exec tail -f /dev/null
 fi

images/warp10/s6/services.d/warp10/run

@@ -8,7 +8,7 @@ WARP10_JAR=${WARP10_HOME}/bin/warp10-${WARP10_VERSION}.jar
 WARP10_CLASS=io.warp10.standalone.Warp
 WARP10_CP="${WARP10_HOME}/etc:${WARP10_JAR}:${WARP10_HOME}/lib/*"
 WARP10_CONFIG_DIR="$WARP10_DATA_DIR/conf"
-CONFIG_FILES="$(find ${WARP10_CONFIG_DIR} -not -path "*/\.*" -name "*.conf" | sort | tr '\n' ' ' 2> /dev/null)"
+CONFIG_FILES="$(find ${WARP10_CONFIG_DIR} -not -path '*/.*' -name '*.conf' | sort | tr '\n' ' ' 2> /dev/null)"
 LOG4J_CONF=${WARP10_HOME}/etc/log4j.properties
 
 if [ -z "$WARP10_HEAP" ]; then
@@ -16,10 +16,10 @@ if [ -z "$WARP10_HEAP" ]; then
 fi
 
 if [ -z "$WARP10_HEAP_MAX" ]; then
-    WARP10_HEAP_MAX=1g
+    WARP10_HEAP_MAX=4g
 fi
 
-JAVA_OPTS="-Djava.awt.headless=true -Xms${WARP10_HEAP} -Xmx${WARP10_HEAP_MAX} -XX:+UseG1GC ${JAVA_OPTS}"
+JAVA_OPTS="-Dlog4j.configuration=file:${LOG4J_CONF} ${JAVA__EXTRA_OPTS} -Djava.awt.headless=true -Xms${WARP10_HEAP} -Xmx${WARP10_HEAP_MAX} -XX:+UseG1GC"
 
 SENSISION_OPTS=
 if [ -n "$ENABLE_SENSISION" ]; then
@@ -28,10 +28,16 @@ if [ -n "$ENABLE_SENSISION" ]; then
     if [ -n "$SENSISION_LABELS" ]; then
         _SENSISION_LABELS="-Dsensision.default.labels=$SENSISION_LABELS"
     fi
-    SENSISION_OPTS="-Dsensision.server.port=0 ${_SENSISION_LABELS} -Dsensision.events.dir=/var/run/sensision/metrics -Dfile.encoding=UTF-8"
+    SENSISION_OPTS="${_SENSISION_LABELS} -Dsensision.events.dir=/var/run/sensision/metrics -Dfile.encoding=UTF-8 ${SENSISION_EXTRA_OPTS}"
 fi
 
-WARP10_CMD="${JAVA} -Dlog4j.configuration=file:${LOG4J_CONF} ${JAVA_OPTS} ${SENSISION_OPTS} -cp ${WARP10_CP} ${WARP10_CLASS} ${CONFIG_FILES}"
+JMX_EXPORTER_OPTS=
+if [ -n "$ENABLE_JMX_EXPORTER" ]; then
+    JMX_EXPORTER_OPTS="-javaagent:/opt/jmx_prom_agent.jar=4803:/opt/jmx_prom.yaml ${JMX_EXPORTER_EXTRA_OPTS}"
+    echo "Starting jmx exporter with Warp 10."
+fi
+
+WARP10_CMD="${JAVA} ${JMX_EXPORTER_OPTS} ${JAVA_OPTS} ${SENSISION_OPTS} -cp ${WARP10_CP} ${WARP10_CLASS} ${CONFIG_FILES}"
 
 echo "Starting Warp 10 with $WARP10_CMD ..."
 exec $WARP10_CMD | tee -a ${WARP10_HOME}/logs/warp10.log

index.js

@@ -1,5 +1,4 @@
 /* eslint-disable global-require */
-
 // eslint-disable-line strict
 let toExport;
 

lib/Config.js

@@ -1,35 +1,13 @@
 /* eslint-disable no-bitwise */
 const assert = require('assert');
 const fs = require('fs');
-const path = require('path');
 
 /**
  * Reads from a config file and returns the content as a config object
  */
 class Config {
-    constructor() {
-        /*
-         * By default, the config file is "config.json" at the root.
-         * It can be overridden using the UTAPI_CONFIG_FILE environment var.
-         */
-        this._basePath = path.resolve(__dirname, '..');
-        this.path = `${this._basePath}/config.json`;
-        if (process.env.UTAPI_CONFIG_FILE !== undefined) {
-            this.path = process.env.UTAPI_CONFIG_FILE;
-        }
-
-        // Read config automatically
-        this._getConfig();
-    }
-
-    _getConfig() {
-        let config;
-        try {
-            const data = fs.readFileSync(this.path, { encoding: 'utf-8' });
-            config = JSON.parse(data);
-        } catch (err) {
-            throw new Error(`could not parse config file: ${err.message}`);
-        }
+    constructor(config) {
+        this.component = config.component;
 
         this.port = 9500;
         if (config.port !== undefined) {
@@ -115,18 +93,26 @@ class Config {
             }
         }
 
-        this.vaultd = {};
-        if (config.vaultd) {
-            if (config.vaultd.port !== undefined) {
-                assert(Number.isInteger(config.vaultd.port)
-                && config.vaultd.port > 0,
-                'bad config: vaultd port must be a positive integer');
-                this.vaultd.port = config.vaultd.port;
-            }
-            if (config.vaultd.host !== undefined) {
-                assert.strictEqual(typeof config.vaultd.host, 'string',
-                    'bad config: vaultd host must be a string');
-                this.vaultd.host = config.vaultd.host;
+        if (config.vaultclient) {
+            // Instance passed from outside
+            this.vaultclient = config.vaultclient;
+            this.vaultd = null;
+        } else {
+            // Connection data
+            this.vaultclient = null;
+            this.vaultd = {};
+            if (config.vaultd) {
+                if (config.vaultd.port !== undefined) {
+                    assert(Number.isInteger(config.vaultd.port)
+                    && config.vaultd.port > 0,
+                    'bad config: vaultd port must be a positive integer');
+                    this.vaultd.port = config.vaultd.port;
+                }
+                if (config.vaultd.host !== undefined) {
+                    assert.strictEqual(typeof config.vaultd.host, 'string',
+                        'bad config: vaultd host must be a string');
+                    this.vaultd.host = config.vaultd.host;
+                }
             }
         }
 
@@ -141,12 +127,11 @@ class Config {
         const { key, cert, ca } = config.certFilePaths
             ? config.certFilePaths : {};
         if (key && cert) {
-            const keypath = (key[0] === '/') ? key : `${this._basePath}/${key}`;
-            const certpath = (cert[0] === '/')
-                ? cert : `${this._basePath}/${cert}`;
+            const keypath = key;
+            const certpath = cert;
             let capath;
             if (ca) {
-                capath = (ca[0] === '/') ? ca : `${this._basePath}/${ca}`;
+                capath = ca;
                 assert.doesNotThrow(() => fs.accessSync(capath, fs.F_OK | fs.R_OK),
                     `File not found or unreachable: ${capath}`);
             }
@@ -172,8 +157,13 @@ class Config {
             + 'expireMetrics must be a boolean');
             this.expireMetrics = config.expireMetrics;
         }
-        return config;
+
+        if (config.onlyCountLatestWhenObjectLocked !== undefined) {
+            assert(typeof config.onlyCountLatestWhenObjectLocked === 'boolean',
+                'bad config: onlyCountLatestWhenObjectLocked must be a boolean');
+            this.onlyCountLatestWhenObjectLocked = config.onlyCountLatestWhenObjectLocked;
+        }
     }
 }
 
-module.exports = new Config();
+module.exports = Config;

lib/Datastore.js

@@ -81,6 +81,17 @@ class Datastore {
         return this._client.call((backend, done) => backend.incr(key, done), cb);
     }
 
+    /**
+     * increment value of a key by the provided value
+     * @param {string} key - key holding the value
+     * @param {string} value - value containing the data
+     * @param {callback} cb - callback
+     * @return {undefined}
+     */
+    incrby(key, value, cb) {
+        return this._client.call((backend, done) => backend.incrby(key, value, done), cb);
+    }
+
     /**
     * decrement value of a key by 1
     * @param {string} key - key holding the value

lib/ListMetrics.js

@@ -6,8 +6,6 @@ const async = require('async');
 const { errors } = require('arsenal');
 const { getMetricFromKey, getKeys, generateStateKey } = require('./schema');
 const s3metricResponseJSON = require('../models/s3metricResponse');
-const config = require('./Config');
-const Vault = require('./Vault');
 
 const MAX_RANGE_MS = (((1000 * 60) * 60) * 24) * 30; // One month.
 
@@ -23,7 +21,6 @@ class ListMetrics {
     constructor(metric, component) {
         this.metric = metric;
         this.service = component;
-        this.vault = new Vault(config);
     }
 
     /**
@@ -83,9 +80,10 @@ class ListMetrics {
         const resources = validator.get(this.metric);
         const timeRange = validator.get('timeRange');
         const datastore = utapiRequest.getDatastore();
+        const vault = utapiRequest.getVault();
         // map account ids to canonical ids
         if (this.metric === 'accounts') {
-            return this.vault.getCanonicalIds(resources, log, (err, list) => {
+            return vault.getCanonicalIds(resources, log, (err, list) => {
                 if (err) {
                     return cb(err);
                 }
@@ -124,7 +122,28 @@ class ListMetrics {
         const fifteenMinutes = 15 * 60 * 1000; // In milliseconds
         const timeRange = [start - fifteenMinutes, end];
         const datastore = utapiRequest.getDatastore();
-        async.mapLimit(resources, 5, (resource, next) => this.getMetrics(resource, timeRange, datastore, log,
+        const vault = utapiRequest.getVault();
+
+        // map account ids to canonical ids
+        if (this.metric === 'accounts') {
+            return vault.getCanonicalIds(resources, log, (err, list) => {
+                if (err) {
+                    return cb(err);
+                }
+                return async.mapLimit(list.message.body, 5,
+                    (item, next) => this.getMetrics(item.canonicalId, timeRange,
+                        datastore, log, (err, res) => {
+                            if (err) {
+                                return next(err);
+                            }
+                            return next(null, Object.assign({}, res,
+                                { accountId: item.accountId }));
+                        }),
+                    cb);
+            });
+        }
+
+        return async.mapLimit(resources, 5, (resource, next) => this.getMetrics(resource, timeRange, datastore, log,
             next), cb);
     }
 
@@ -293,11 +312,10 @@ class ListMetrics {
             });
 
             if (!areMetricsPositive) {
-                return cb(errors.InternalError.customizeDescription(
-                    'Utapi is in a transient state for this time period as '
-                    + 'metrics are being collected. Please try again in a few '
-                    + 'minutes.',
-                ));
+                log.info('negative metric value found', {
+                    error: resource,
+                    method: 'ListMetrics.getMetrics',
+                });
             }
             /**
             * Batch result is of the format

lib/UtapiClient.js

@@ -63,7 +63,6 @@ const methods = {
     getObjectTagging: { method: '_genericPushMetric', changesData: false },
     putObject: { method: '_genericPushMetricPutObject', changesData: true },
     copyObject: { method: '_genericPushMetricPutObject', changesData: true },
-    putData: { method: '_genericPushMetricPutObject', changesData: true },
     putObjectAcl: { method: '_genericPushMetric', changesData: true },
     putObjectLegalHold: { method: '_genericPushMetric', changesData: true },
     putObjectRetention: { method: '_genericPushMetric', changesData: true },
@@ -91,12 +90,16 @@ const methods = {
     },
     putBucketObjectLock: { method: '_genericPushMetric', changesData: true },
     getBucketObjectLock: { method: '_genericPushMetric', changesData: true },
+    replicateObject: { method: '_genericPushMetricPutObject', changesData: true },
+    replicateTags: { method: '_genericPushMetric', changesData: true },
+    replicateDelete: { method: '_pushMetricDeleteMarkerObject', changesData: true },
 };
 
 const metricObj = {
     buckets: 'bucket',
     accounts: 'accountId',
     users: 'userId',
+    location: 'location',
 };
 
 class UtapiClient {
@@ -120,13 +123,17 @@ class UtapiClient {
         const api = (config || {}).logApi || werelogs;
         this.log = new api.Logger('UtapiClient');
         // By default, we push all resource types
-        this.metrics = ['buckets', 'accounts', 'users', 'service'];
+        this.metrics = ['buckets', 'accounts', 'users', 'service', 'location'];
         this.service = 's3';
         this.disableOperationCounters = false;
         this.enabledOperationCounters = [];
         this.disableClient = true;
 
-        if (config) {
+        if (config && !config.disableClient) {
+            this.disableClient = false;
+            this.expireMetrics = config.expireMetrics;
+            this.expireMetricsTTL = config.expireMetricsTTL || 0;
+
             if (config.metrics) {
                 const message = 'invalid property in UtapiClient configuration';
                 assert(Array.isArray(config.metrics), `${message}: metrics `
@@ -154,9 +161,6 @@ class UtapiClient {
             if (config.enabledOperationCounters) {
                 this.enabledOperationCounters = config.enabledOperationCounters;
             }
-            this.disableClient = false;
-            this.expireMetrics = config.expireMetrics;
-            this.expireMetricsTTL = config.expireMetricsTTL || 0;
         }
     }
 
@@ -540,10 +544,13 @@ class UtapiClient {
         const paramsArr = this._getParamsArr(params);
         paramsArr.forEach(p => {
             cmds.push(['incr', generateCounter(p, 'numberOfObjectsCounter')]);
-            if (this._isCounterEnabled('deleteObject')) {
-                cmds.push(['incr', generateKey(p, 'deleteObject', timestamp)]);
+            const counterAction = action === 'putDeleteMarkerObject' ? 'deleteObject' : action;
+            if (this._isCounterEnabled(counterAction)) {
+                cmds.push(['incr', generateKey(p, counterAction, timestamp)]);
             }
+            cmds.push(['zrangebyscore', generateStateKey(p, 'storageUtilized'), timestamp, timestamp]);
         });
+
         return this.ds.batch(cmds, (err, results) => {
             if (err) {
                 log.error('error pushing metric', {
@@ -577,13 +584,48 @@ class UtapiClient {
                 // empty.
                 actionCounter = Number.isNaN(actionCounter)
                     || actionCounter < 0 ? 1 : actionCounter;
+
+                if (Number.isInteger(params.byteLength)) {
+                    /* byteLength is passed in from cloudserver under the follow conditions:
+                     * - bucket versioning is suspended
+                     * - object version id is null
+                     * - the content length of the object exists
+                     * In this case, the master key is deleted and replaced with a delete marker.
+                     * The decrement accounts for the deletion of the master key when utapi reports
+                     * on the number of objects.
+                     */
+                    actionCounter -= 1;
+                }
+
                 const key = generateStateKey(p, 'numberOfObjects');
+
+                const byteArr = results[index + commandsGroupSize - 1][1];
+                const oldByteLength = byteArr ? parseInt(byteArr[0], 10) : 0;
+                const newByteLength = member.serialize(Math.max(0, oldByteLength - params.byteLength));
+
                 cmds2.push(
                     ['zremrangebyscore', key, timestamp, timestamp],
                     ['zadd', key, timestamp, member.serialize(actionCounter)],
+
                 );
+
+                if (Number.isInteger(params.byteLength)) {
+                    cmds2.push(
+                        ['decr', generateCounter(p, 'numberOfObjectsCounter')],
+                        ['decrby', generateCounter(p, 'storageUtilizedCounter'), params.byteLength],
+                    );
+                }
+
+                if (byteArr) {
+                    cmds2.push(
+                        ['zremrangebyscore', generateStateKey(p, 'storageUtilized'), timestamp, timestamp],
+                        ['zadd', generateStateKey(p, 'storageUtilized'), timestamp, newByteLength],
+                    );
+                }
+
                 return true;
             });
+
             if (noErr) {
                 return this.ds.batch(cmds2, cb);
             }
@@ -1025,10 +1067,10 @@ class UtapiClient {
                     storageUtilizedDelta],
                 [redisCmd, generateCounter(p, 'numberOfObjectsCounter')],
             );
-            if (action !== 'putData' && this._isCounterEnabled(action)) {
+            if (this._isCounterEnabled(action)) {
                 cmds.push(['incr', generateKey(p, action, timestamp)]);
             }
-            if (action === 'putObject' || action === 'putData') {
+            if (action === 'putObject' || action === 'replicateObject') {
                 cmds.push(
                     ['incrby', generateKey(p, 'incomingBytes', timestamp),
                         newByteLength],
@@ -1114,6 +1156,69 @@ class UtapiClient {
         });
     }
 
+    /**
+     *
+     * @param {string} location - name of data location
+     * @param {number} updateSize - size in bytes to update location metric by,
+     * could be negative, indicating deleted object
+     * @param {string} reqUid - Request Unique Identifier
+     * @param {function} callback - callback to call
+     * @return {undefined}
+     */
+    pushLocationMetric(location, updateSize, reqUid, callback) {
+        const log = this.log.newRequestLoggerFromSerializedUids(reqUid);
+        const params = {
+            level: 'location',
+            service: 's3',
+            location,
+        };
+        this._checkMetricTypes(params);
+        const action = (updateSize < 0) ? 'decrby' : 'incrby';
+        const size = (updateSize < 0) ? -updateSize : updateSize;
+        return this.ds[action](generateKey(params, 'locationStorage'), size,
+            err => {
+                if (err) {
+                    log.error('error pushing metric', {
+                        method: 'UtapiClient.pushLocationMetric',
+                        error: err,
+                    });
+                    return callback(errors.InternalError);
+                }
+                return callback();
+            });
+    }
+
+    /**
+     *
+     * @param {string} location - name of data backend to get metric for
+     * @param {string} reqUid - Request Unique Identifier
+     * @param {function} callback - callback to call
+     * @return {undefined}
+     */
+    getLocationMetric(location, reqUid, callback) {
+        const log = this.log.newRequestLoggerFromSerializedUids(reqUid);
+        const params = {
+            level: 'location',
+            service: 's3',
+            location,
+        };
+        const redisKey = generateKey(params, 'locationStorage');
+        return this.ds.get(redisKey, (err, bytesStored) => {
+            if (err) {
+                log.error('error getting metric', {
+                    method: 'UtapiClient: getLocationMetric',
+                    error: err,
+                });
+                return callback(errors.InternalError);
+            }
+            // if err and bytesStored are null, key does not exist yet
+            if (bytesStored === null) {
+                return callback(null, 0);
+            }
+            return callback(null, bytesStored);
+        });
+    }
+
     /**
      * Get storage used by bucket/account/user/service
      * @param {object} params - params for the metrics

lib/UtapiReindex.js

@@ -12,16 +12,23 @@ const RedisClient = require('../libV2/redis');
 const REINDEX_SCHEDULE = '0 0 * * Sun';
 const REINDEX_LOCK_KEY = 's3:utapireindex:lock';
 const REINDEX_LOCK_TTL = (60 * 60) * 24;
+const REINDEX_PYTHON_INTERPRETER = process.env.REINDEX_PYTHON_INTERPRETER !== undefined
+    ? process.env.REINDEX_PYTHON_INTERPRETER
+    : 'python3.7';
+
+const EXIT_CODE_SENTINEL_CONNECTION = 100;
 
 class UtapiReindex {
     constructor(config) {
         this._enabled = false;
         this._schedule = REINDEX_SCHEDULE;
-        this._sentinel = {
-            host: '127.0.0.1',
-            port: 16379,
+        this._redis = {
             name: 'scality-s3',
             sentinelPassword: '',
+            sentinels: [{
+                host: '127.0.0.1',
+                port: 16379,
+            }],
         };
         this._bucketd = {
             host: '127.0.0.1',
@@ -39,14 +46,13 @@ class UtapiReindex {
         if (config && config.password) {
             this._password = config.password;
         }
-        if (config && config.sentinel) {
+        if (config && config.redis) {
             const {
-                host, port, name, sentinelPassword,
-            } = config.sentinel;
-            this._sentinel.host = host || this._sentinel.host;
-            this._sentinel.port = port || this._sentinel.port;
-            this._sentinel.name = name || this._sentinel.name;
-            this._sentinel.sentinelPassword = sentinelPassword || this._sentinel.sentinelPassword;
+                name, sentinelPassword, sentinels,
+            } = config.redis;
+            this._redis.name = name || this._redis.name;
+            this._redis.sentinelPassword = sentinelPassword || this._redis.sentinelPassword;
+            this._redis.sentinels = sentinels || this._redis.sentinels;
         }
         if (config && config.bucketd) {
             const { host, port } = config.bucketd;
@@ -58,17 +64,16 @@ class UtapiReindex {
             this._log = new werelogs.Logger('UtapiReindex', { level, dump });
         }
 
+        this._onlyCountLatestWhenObjectLocked = (config && config.onlyCountLatestWhenObjectLocked === true);
+
         this._requestLogger = this._log.newRequestLogger();
     }
 
     _getRedisClient() {
         const client = new RedisClient({
-            sentinels: [{
-                host: this._sentinel.host,
-                port: this._sentinel.port,
-            }],
-            name: this._sentinel.name,
-            sentinelPassword: this._sentinel.sentinelPassword,
+            sentinels: this._redis.sentinels,
+            name: this._redis.name,
+            sentinelPassword: this._redis.sentinelPassword,
             password: this._password,
         });
         client.connect();
@@ -83,17 +88,18 @@ class UtapiReindex {
         return this.ds.del(REINDEX_LOCK_KEY);
     }
 
-    _buildFlags() {
+    _buildFlags(sentinel) {
         const flags = {
             /* eslint-disable camelcase */
-            sentinel_ip: this._sentinel.host,
-            sentinel_port: this._sentinel.port,
-            sentinel_cluster_name: this._sentinel.name,
+            sentinel_ip: sentinel.host,
+            sentinel_port: sentinel.port,
+            sentinel_cluster_name: this._redis.name,
             bucketd_addr: `http://${this._bucketd.host}:${this._bucketd.port}`,
         };
-        if (this._sentinel.sentinelPassword) {
-            flags.redis_password = this._sentinel.sentinelPassword;
+        if (this._redis.sentinelPassword) {
+            flags.redis_password = this._redis.sentinelPassword;
         }
+
         /* eslint-enable camelcase */
         const opts = [];
         Object.keys(flags)
@@ -102,17 +108,17 @@ class UtapiReindex {
                 opts.push(name);
                 opts.push(flags[flag]);
             });
+
+        if (this._onlyCountLatestWhenObjectLocked) {
+            opts.push('--only-latest-when-locked');
+        }
         return opts;
     }
 
-    _runScript(path, done) {
-        const flags = this._buildFlags();
-        this._requestLogger.debug(`launching subprocess ${path} `
-            + `with flags: ${flags}`);
-        const process = childProcess.spawn('python3.4', [
-            path,
-            ...flags,
-        ]);
+    _runScriptWithSentinels(path, remainingSentinels, done) {
+        const flags = this._buildFlags(remainingSentinels.shift());
+        this._requestLogger.debug(`launching subprocess ${path} with flags: ${flags}`);
+        const process = childProcess.spawn(REINDEX_PYTHON_INTERPRETER, [path, ...flags]);
         process.stdout.on('data', data => {
             this._requestLogger.info('received output from script', {
                 output: Buffer.from(data).toString(),
@@ -137,6 +143,17 @@ class UtapiReindex {
                     statusCode: code,
                     script: path,
                 });
+                if (code === EXIT_CODE_SENTINEL_CONNECTION) {
+                    if (remainingSentinels.length > 0) {
+                        this._requestLogger.info('retrying with next sentinel host', {
+                            script: path,
+                        });
+                        return this._runScriptWithSentinels(path, remainingSentinels, done);
+                    }
+                    this._requestLogger.error('no more sentinel host to try', {
+                        script: path,
+                    });
+                }
             } else {
                 this._requestLogger.info('script exited successfully', {
                     statusCode: code,
@@ -147,6 +164,11 @@ class UtapiReindex {
         });
     }
 
+    _runScript(path, done) {
+        const remainingSentinels = [...this._redis.sentinels];
+        this._runScriptWithSentinels(path, remainingSentinels, done);
+    }
+
     _attemptLock(job) {
         this._requestLogger.info('attempting to acquire the lock to begin job');
         this._lock()

lib/UtapiRequest.js

@@ -14,6 +14,15 @@ class UtapiRequest {
         this._datastore = null;
         this._requestQuery = null;
         this._requestPath = null;
+        this._vault = null;
+    }
+
+    getVault() {
+        return this._vault;
+    }
+
+    setVault() {
+        return this._vault;
     }
 
     /**

lib/reindex/reporting.py

@@ -1,16 +1,21 @@
-import requests
-import redis
-import json
+import argparse
 import ast
-import sys
-import time
-import urllib
+from concurrent.futures import ThreadPoolExecutor
+import json
+import logging
 import re
+import redis
+import requests
 import sys
 from threading import Thread
-from concurrent.futures import ThreadPoolExecutor
+import time
+import urllib
 
-import argparse
+logging.basicConfig(level=logging.INFO)
+_log = logging.getLogger('utapi-reindex:reporting')
+
+SENTINEL_CONNECT_TIMEOUT_SECONDS = 10
+EXIT_CODE_SENTINEL_CONNECTION_ERROR = 100
 
 def get_options():
     parser = argparse.ArgumentParser()
@@ -29,8 +34,19 @@ class askRedis():
 
     def __init__(self, ip="127.0.0.1", port="16379", sentinel_cluster_name="scality-s3", password=None):
         self._password = password
-        r = redis.Redis(host=ip, port=port, db=0, password=password)
-        self._ip, self._port = r.sentinel_get_master_addr_by_name(sentinel_cluster_name)
+        r = redis.Redis(
+            host=ip,
+            port=port,
+            db=0,
+            password=password,
+            socket_connect_timeout=SENTINEL_CONNECT_TIMEOUT_SECONDS
+        )
+        try:
+            self._ip, self._port = r.sentinel_get_master_addr_by_name(sentinel_cluster_name)
+        except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError) as e:
+            _log.error(f'Failed to connect to redis sentinel at {ip}:{port}: {e}')
+            # use a specific error code to hint on retrying with another sentinel node
+            sys.exit(EXIT_CODE_SENTINEL_CONNECTION_ERROR)
 
     def read(self, resource, name):
         r = redis.Redis(host=self._ip, port=self._port, db=0, password=self._password)

lib/reindex/s3_bucketd.py

@@ -1,5 +1,6 @@
 import argparse
 import concurrent.futures as futures
+import functools
 import itertools
 import json
 import logging
@@ -8,9 +9,9 @@ import re
 import sys
 import time
 import urllib
+from pathlib import Path
 from collections import defaultdict, namedtuple
 from concurrent.futures import ThreadPoolExecutor
-from pprint import pprint
 
 import redis
 import requests
@@ -24,6 +25,9 @@ MPU_SHADOW_BUCKET_PREFIX = 'mpuShadowBucket'
 
 ACCOUNT_UPDATE_CHUNKSIZE = 100
 
+SENTINEL_CONNECT_TIMEOUT_SECONDS = 10
+EXIT_CODE_SENTINEL_CONNECTION_ERROR = 100
+
 def get_options():
     parser = argparse.ArgumentParser()
     parser.add_argument("-i", "--sentinel-ip", default='127.0.0.1', help="Sentinel IP")
@@ -31,9 +35,39 @@ def get_options():
     parser.add_argument("-v", "--redis-password", default=None, help="Redis AUTH Password")
     parser.add_argument("-n", "--sentinel-cluster-name", default='scality-s3', help="Redis cluster name")
     parser.add_argument("-s", "--bucketd-addr", default='http://127.0.0.1:9000', help="URL of the bucketd server")
-    parser.add_argument("-w", "--worker", default=10, help="Number of workers")
-    parser.add_argument("-b", "--bucket", default=False, help="Bucket to be processed")
-    return parser.parse_args()
+    parser.add_argument("-w", "--worker", default=10, type=int, help="Number of workers")
+    parser.add_argument("-r", "--max-retries", default=2, type=int, help="Max retries before failing a bucketd request")
+    parser.add_argument("--only-latest-when-locked", action='store_true', help="Only index the latest version of a key when the bucket has a default object lock policy")
+    parser.add_argument("--debug", action='store_true', help="Enable debug logging")
+    parser.add_argument("--dry-run", action="store_true", help="Do not update redis")
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument("-a", "--account", default=[], help="account canonical ID (all account buckets will be processed)", action="append", type=nonempty_string('account'))
+    group.add_argument("--account-file", default=None, help="file containing account canonical IDs, one ID per line", type=existing_file)
+    group.add_argument("-b", "--bucket", default=[], help="bucket name", action="append", type=nonempty_string('bucket'))
+    group.add_argument("--bucket-file", default=None, help="file containing bucket names, one bucket name per line", type=existing_file)
+
+    options = parser.parse_args()
+    if options.bucket_file:
+        with open(options.bucket_file) as f:
+            options.bucket = [line.strip() for line in f if line.strip()]
+    elif options.account_file:
+        with open(options.account_file) as f:
+            options.account = [line.strip() for line in f if line.strip()]
+
+    return options
+
+def nonempty_string(flag):
+    def inner(value):
+        if not value.strip():
+            raise argparse.ArgumentTypeError("%s: value must not be empty"%flag)
+        return value
+    return inner
+
+def existing_file(path):
+    path = Path(path).resolve()
+    if not path.exists():
+        raise argparse.ArgumentTypeError("File does not exist: %s"%path)
+    return path
 
 def chunks(iterable, size):
     it = iter(iterable)
@@ -48,22 +82,38 @@ def _encoded(func):
         return urllib.parse.quote(val.encode('utf-8'))
     return inner
 
-Bucket = namedtuple('Bucket', ['userid', 'name'])
+Bucket = namedtuple('Bucket', ['userid', 'name', 'object_lock_enabled'])
 MPU = namedtuple('MPU', ['bucket', 'key', 'upload_id'])
 BucketContents = namedtuple('BucketContents', ['bucket', 'obj_count', 'total_size'])
 
+class MaxRetriesReached(Exception):
+    def __init__(self, url):
+        super().__init__('Max retries reached for request to %s'%url)
+
+class InvalidListing(Exception):
+    def __init__(self, bucket):
+        super().__init__('Invalid contents found while listing bucket %s'%bucket)
+
+class BucketNotFound(Exception):
+    def __init__(self, bucket):
+        super().__init__('Bucket %s not found'%bucket)
+
 class BucketDClient:
 
     '''Performs Listing calls against bucketd'''
-    __url_format = '{addr}/default/bucket/{bucket}'
+    __url_attribute_format = '{addr}/default/attributes/{bucket}'
+    __url_bucket_format = '{addr}/default/bucket/{bucket}'
     __headers = {"x-scal-request-uids": "utapi-reindex-list-buckets"}
 
-    def __init__(self, bucketd_addr=None):
+    def __init__(self, bucketd_addr=None, max_retries=2, only_latest_when_locked=False):
         self._bucketd_addr = bucketd_addr
+        self._max_retries = max_retries
+        self._only_latest_when_locked = only_latest_when_locked
         self._session = requests.Session()
 
     def _do_req(self, url, check_500=True, **kwargs):
-        while True:
+        # Add 1 for the initial request
+        for x in range(self._max_retries + 1):
             try:
                 resp = self._session.get(url, timeout=30, verify=False, headers=self.__headers, **kwargs)
                 if check_500 and resp.status_code == 500:
@@ -76,6 +126,8 @@ class BucketDClient:
                 _log.error('Error during listing, sleeping 5 secs %s'%url)
                 time.sleep(5)
 
+        raise MaxRetriesReached(url)
+
     def _list_bucket(self, bucket, **kwargs):
         '''
         Lists a bucket lazily until "empty"
@@ -88,7 +140,7 @@ class BucketDClient:
         parameters value. On the first request the function will be called with
         `None` and should return its initial value. Return `None` for the param to be excluded.
         '''
-        url = self.__url_format.format(addr=self._bucketd_addr, bucket=bucket)
+        url = self.__url_bucket_format.format(addr=self._bucketd_addr, bucket=bucket)
         static_params = {k: v for k, v in kwargs.items() if not callable(v)}
         dynamic_params = {k: v for k, v in kwargs.items() if callable(v)}
         is_truncated = True # Set to True for first loop
@@ -101,6 +153,9 @@ class BucketDClient:
                 _log.debug('listing bucket bucket: %s params: %s'%(
                     bucket, ', '.join('%s=%s'%p for p in params.items())))
                 resp = self._do_req(url, params=params)
+                if resp.status_code == 404:
+                    _log.debug('Bucket not found bucket: %s'%bucket)
+                    return
                 if resp.status_code == 200:
                     payload = resp.json()
             except ValueError as e:
@@ -108,6 +163,9 @@ class BucketDClient:
                 _log.error('Invalid listing response body! bucket:%s params:%s'%(
                     bucket, ', '.join('%s=%s'%p for p in params.items())))
                 continue
+            except MaxRetriesReached:
+                _log.error('Max retries reached listing bucket:%s'%bucket)
+                raise
             except Exception as e:
                 _log.exception(e)
                 _log.error('Unhandled exception during listing! bucket:%s params:%s'%(
@@ -119,7 +177,37 @@ class BucketDClient:
             else:
                 is_truncated = len(payload) > 0
 
-    def list_buckets(self):
+    @functools.lru_cache(maxsize=16)
+    def _get_bucket_attributes(self, name):
+        url = self.__url_attribute_format.format(addr=self._bucketd_addr, bucket=name)
+        try:
+            resp = self._do_req(url)
+            if resp.status_code == 200:
+                return resp.json()
+            else:
+                _log.error('Error getting bucket attributes bucket:%s status_code:%s'%(name, resp.status_code))
+                raise BucketNotFound(name)
+        except ValueError as e:
+            _log.exception(e)
+            _log.error('Invalid attributes response body! bucket:%s'%name)
+            raise
+        except MaxRetriesReached:
+            _log.error('Max retries reached getting bucket attributes bucket:%s'%name)
+            raise
+        except Exception as e:
+            _log.exception(e)
+            _log.error('Unhandled exception getting bucket attributes bucket:%s'%name)
+            raise
+
+    def get_bucket_md(self, name):
+        md = self._get_bucket_attributes(name)
+        canonId = md.get('owner')
+        if canonId is None:
+            _log.error('No owner found for bucket %s'%name)
+            raise InvalidListing(name)
+        return Bucket(canonId, name, md.get('objectLockEnabled', False))
+
+    def list_buckets(self, account=None):
 
         def get_next_marker(p):
             if p is None:
@@ -131,13 +219,24 @@ class BucketDClient:
             'maxKeys': 1000,
             'marker': get_next_marker
         }
+
+        if account is not None:
+            params['prefix'] = '%s..|..' % account
+
         for _, payload in self._list_bucket(USERS_BUCKET, **params):
             buckets = []
-            for result in payload['Contents']:
+            for result in payload.get('Contents', []):
                 match = re.match("(\w+)..\|..(\w+.*)", result['key'])
-                buckets.append(Bucket(*match.groups()))
-            yield buckets
+                bucket = Bucket(*match.groups(), False)
+                # We need to get the attributes for each bucket to determine if it is locked
+                if self._only_latest_when_locked:
+                    bucket_attrs = self._get_bucket_attributes(bucket.name)
+                    object_lock_enabled = bucket_attrs.get('objectLockEnabled', False)
+                    bucket = bucket._replace(object_lock_enabled=object_lock_enabled)
+                buckets.append(bucket)
 
+            if buckets:
+                yield buckets
 
     def list_mpus(self, bucket):
         _bucket = MPU_SHADOW_BUCKET_PREFIX + bucket.name
@@ -174,15 +273,12 @@ class BucketDClient:
                     upload_id=key['value']['UploadId']))
         return keys
 
-    def _sum_objects(self, listing):
+    def _sum_objects(self, bucket, listing, only_latest_when_locked = False):
         count = 0
         total_size = 0
-        last_master = None
-        last_size = None
-        for _, payload in listing:
-            contents = payload['Contents'] if isinstance(payload, dict) else payload
-            for obj in contents:
-                count += 1
+        last_key = None
+        try:
+            for obj in listing:
                 if isinstance(obj['value'], dict):
                     # bucketd v6 returns a dict:
                     data = obj.get('value', {})
@@ -190,40 +286,52 @@ class BucketDClient:
                 else:
                     # bucketd v7 returns an encoded string
                     data = json.loads(obj['value'])
-                    size = data["content-length"]
+                    size = data.get('content-length', 0)
+
+                is_latest = obj['key'] != last_key
+                last_key = obj['key']
+
+                if only_latest_when_locked and bucket.object_lock_enabled and not is_latest:
+                    _log.debug('Skipping versioned key: %s'%obj['key'])
+                    continue
+
+                count += 1
                 total_size += size
 
-                # If versioned, subtract the size of the master to avoid double counting
-                if last_master is not None and obj['key'].startswith(last_master + '\x00'):
-                    _log.info('Detected versioned key: %s - subtracting master size: %i'% (
-                        obj['key'],
-                        last_size,
-                    ))
-                    total_size -= last_size
-                    count -= 1
-                    last_master = None
-
-                # Only save master versions
-                elif '\x00' not in obj['key']:
-                    last_master = obj['key']
-                    last_size = size
-
+        except InvalidListing:
+            _log.error('Invalid contents in listing. bucket:%s'%bucket.name)
+            raise InvalidListing(bucket.name)
         return count, total_size
 
+    def _extract_listing(self, key, listing):
+        for status_code, payload in listing:
+            contents = payload[key] if isinstance(payload, dict) else payload
+            if contents is None:
+                raise InvalidListing('')
+            for obj in contents:
+                yield obj
+
     def count_bucket_contents(self, bucket):
 
-        def get_next_marker(p):
-            if p is None or len(p) == 0:
+        def get_key_marker(p):
+            if p is None:
                 return ''
-            return p[-1].get('key', '')
+            return p.get('NextKeyMarker', '')
+
+        def get_vid_marker(p):
+            if p is None:
+                return ''
+            return p.get('NextVersionIdMarker', '')
 
         params = {
-            'listingType': 'Basic',
+            'listingType': 'DelimiterVersions',
             'maxKeys': 1000,
-            'gt': get_next_marker,
+            'keyMarker': get_key_marker,
+            'versionIdMarker': get_vid_marker,
         }
 
-        count, total_size = self._sum_objects(self._list_bucket(bucket.name, **params))
+        listing = self._list_bucket(bucket.name, **params)
+        count, total_size = self._sum_objects(bucket, self._extract_listing('Versions', listing), self._only_latest_when_locked)
         return BucketContents(
             bucket=bucket,
             obj_count=count,
@@ -231,7 +339,8 @@ class BucketDClient:
         )
 
     def count_mpu_parts(self, mpu):
-        _bucket = MPU_SHADOW_BUCKET_PREFIX + mpu.bucket.name
+        shadow_bucket_name = MPU_SHADOW_BUCKET_PREFIX + mpu.bucket.name
+        shadow_bucket = mpu.bucket._replace(name=shadow_bucket_name)
 
         def get_prefix(p):
             if p is None:
@@ -251,30 +360,53 @@ class BucketDClient:
             'listingType': 'Delimiter',
         }
 
-        count, total_size = self._sum_objects(self._list_bucket(_bucket, **params))
+        listing = self._list_bucket(shadow_bucket_name, **params)
+        count, total_size = self._sum_objects(shadow_bucket, self._extract_listing('Contents', listing))
         return BucketContents(
-            bucket=mpu.bucket._replace(name=_bucket),
+            bucket=shadow_bucket,
             obj_count=0, # MPU parts are not counted towards numberOfObjects
             total_size=total_size
         )
 
+def list_all_buckets(bucket_client):
+    return bucket_client.list_buckets()
+
+def list_specific_accounts(bucket_client, accounts):
+    for account in accounts:
+        yield from bucket_client.list_buckets(account=account)
+
+def list_specific_buckets(bucket_client, buckets):
+    batch = []
+    for bucket in buckets:
+        try:
+            batch.append(bucket_client.get_bucket_md(bucket))
+        except BucketNotFound:
+            _log.error('Failed to list bucket %s. Removing from results.'%bucket)
+            continue
+
+    yield batch
 
 def index_bucket(client, bucket):
     '''
         Takes an instance of BucketDClient and a bucket name, and returns a
         tuple of BucketContents for the passed bucket and its mpu shadow bucket.
     '''
-    bucket_total = client.count_bucket_contents(bucket)
-    mpus = client.list_mpus(bucket)
-    if not mpus:
-        return bucket_total
+    try:
+        bucket_total = client.count_bucket_contents(bucket)
+        mpus = client.list_mpus(bucket)
+        if not mpus:
+            return bucket_total
 
-    total_size = bucket_total.total_size
-    mpu_totals = [client.count_mpu_parts(m) for m in mpus]
-    for mpu in mpu_totals:
-        total_size += mpu.total_size
+        total_size = bucket_total.total_size
+        mpu_totals = [client.count_mpu_parts(m) for m in mpus]
+        for mpu in mpu_totals:
+            total_size += mpu.total_size
 
-    return bucket_total._replace(total_size=total_size)
+        return bucket_total._replace(total_size=total_size)
+    except Exception as e:
+        _log.exception(e)
+        _log.error('Error during listing. Removing from results bucket:%s'%bucket.name)
+        raise InvalidListing(bucket.name)
 
 def update_report(report, key, obj_count, total_size):
     '''Convenience function to update the report dicts'''
@@ -292,9 +424,16 @@ def get_redis_client(options):
         host=options.sentinel_ip,
         port=options.sentinel_port,
         db=0,
-        password=options.redis_password
+        password=options.redis_password,
+        socket_connect_timeout=SENTINEL_CONNECT_TIMEOUT_SECONDS
     )
-    ip, port = sentinel.sentinel_get_master_addr_by_name(options.sentinel_cluster_name)
+    try:
+        ip, port = sentinel.sentinel_get_master_addr_by_name(options.sentinel_cluster_name)
+    except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError) as e:
+        _log.error(f'Failed to connect to redis sentinel at {options.sentinel_ip}:{options.sentinel_port}: {e}')
+        # use a specific error code to hint on retrying with another sentinel node
+        sys.exit(EXIT_CODE_SENTINEL_CONNECTION_ERROR)
+
     return redis.Redis(
         host=ip,
         port=port,
@@ -328,54 +467,120 @@ def log_report(resource, name, obj_count, total_size):
 
 if __name__ == '__main__':
     options = get_options()
-    bucket_client = BucketDClient(options.bucketd_addr)
+    if options.debug:
+        _log.setLevel(logging.DEBUG)
+
+    bucket_client = BucketDClient(options.bucketd_addr, options.max_retries, options.only_latest_when_locked)
     redis_client = get_redis_client(options)
     account_reports = {}
     observed_buckets = set()
+    failed_accounts = set()
+
+    if options.account:
+        batch_generator = list_specific_accounts(bucket_client, options.account)
+    elif options.bucket:
+        batch_generator = list_specific_buckets(bucket_client, options.bucket)
+    else:
+        batch_generator = list_all_buckets(bucket_client)
+
     with ThreadPoolExecutor(max_workers=options.worker) as executor:
-        for batch in bucket_client.list_buckets():
+        for batch in batch_generator:
             bucket_reports = {}
-            jobs = [executor.submit(index_bucket, bucket_client, b) for b in batch]
-            for job in futures.as_completed(jobs):
-                total = job.result() # Summed bucket and shadowbucket totals
+            jobs = { executor.submit(index_bucket, bucket_client, b): b for b in batch }
+            for job in futures.as_completed(jobs.keys()):
+                try:
+                    total = job.result() # Summed bucket and shadowbucket totals
+                except InvalidListing:
+                    _bucket = jobs[job]
+                    _log.error('Failed to list bucket %s. Removing from results.'%_bucket.name)
+                    # Add the bucket to observed_buckets anyway to avoid clearing existing metrics
+                    observed_buckets.add(_bucket.name)
+                    # If we can not list one of an account's buckets we can not update its total
+                    failed_accounts.add(_bucket.userid)
+                    continue
                 observed_buckets.add(total.bucket.name)
                 update_report(bucket_reports, total.bucket.name, total.obj_count, total.total_size)
                 update_report(account_reports, total.bucket.userid, total.obj_count, total.total_size)
 
             # Bucket reports can be updated as we get them
-            pipeline = redis_client.pipeline(transaction=False)  # No transaction to reduce redis load
-            for bucket, report in bucket_reports.items():
-                update_redis(pipeline, 'buckets', bucket, report['obj_count'], report['total_size'])
-                log_report('buckets', bucket, report['obj_count'], report['total_size'])
+            if options.dry_run:
+                for bucket, report in bucket_reports.items():
+                    _log.info(
+                        "DryRun: resource buckets [%s] would be updated with obj_count %i and total_size %i" % (
+                            bucket, report['obj_count'], report['total_size']
+                        )
+                    )
+            else:
+                pipeline = redis_client.pipeline(transaction=False)  # No transaction to reduce redis load
+                for bucket, report in bucket_reports.items():
+                    update_redis(pipeline, 'buckets', bucket, report['obj_count'], report['total_size'])
+                    log_report('buckets', bucket, report['obj_count'], report['total_size'])
+                pipeline.execute()
+
+    stale_buckets = set()
+    recorded_buckets = set(get_resources_from_redis(redis_client, 'buckets'))
+    if options.bucket:
+        stale_buckets = { b for b in options.bucket if b not in observed_buckets }
+    elif options.account:
+        _log.warning('Stale buckets will not be cleared when using the --account or --account-file flags')
+    else:
+        stale_buckets = recorded_buckets.difference(observed_buckets)
+
+    _log.info('Found %s stale buckets' % len(stale_buckets))
+    if options.dry_run:
+        _log.info("DryRun: not updating stale buckets")
+    else:
+        for chunk in chunks(stale_buckets, ACCOUNT_UPDATE_CHUNKSIZE):
+            pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
+            for bucket in chunk:
+                update_redis(pipeline, 'buckets', bucket, 0, 0)
+                log_report('buckets', bucket, 0, 0)
             pipeline.execute()
 
-    # Update total account reports in chunks
-    for chunk in chunks(account_reports.items(), ACCOUNT_UPDATE_CHUNKSIZE):
-        pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
-        for userid, report in chunk:
-            update_redis(pipeline, 'accounts', userid, report['obj_count'], report['total_size'])
-            log_report('accounts', userid, report['obj_count'], report['total_size'])
-        pipeline.execute()
+    # Account metrics are not updated if a bucket is specified
+    if options.bucket:
+        _log.warning('Account metrics will not be updated when using the --bucket or --bucket-file flags')
+    else:
+        # Don't update any accounts with failed listings
+        without_failed = filter(lambda x: x[0] not in failed_accounts, account_reports.items())
+        if options.dry_run:
+            for userid, report in account_reports.items():
+                _log.info(
+                    "DryRun: resource account [%s] would be updated with obj_count %i and total_size %i" % (
+                        userid, report['obj_count'], report['total_size']
+                    )
+                )
+        else:
+            # Update total account reports in chunks
+            for chunk in chunks(without_failed, ACCOUNT_UPDATE_CHUNKSIZE):
+                pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
+                for userid, report in chunk:
+                    update_redis(pipeline, 'accounts', userid, report['obj_count'], report['total_size'])
+                    log_report('accounts', userid, report['obj_count'], report['total_size'])
+                pipeline.execute()
 
-    observed_accounts = set(account_reports.keys())
-    recorded_accounts = set(get_resources_from_redis(redis_client, 'accounts'))
-    recorded_buckets = set(get_resources_from_redis(redis_client, 'buckets'))
+        if options.account:
+            for account in options.account:
+                if account in failed_accounts:
+                    _log.error("No metrics updated for account %s, one or more buckets failed" % account)
 
-    # Stale accounts and buckets are ones that do not appear in the listing, but have recorded values
-    stale_accounts = recorded_accounts.difference(observed_accounts)
-    _log.info('Found %s stale accounts' % len(stale_accounts))
-    for chunk in chunks(stale_accounts, ACCOUNT_UPDATE_CHUNKSIZE):
-        pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
-        for account in chunk:
-            update_redis(pipeline, 'accounts', account, 0, 0)
-            log_report('accounts', account, 0, 0)
-        pipeline.execute()
+        # Include failed_accounts in observed_accounts to avoid clearing metrics
+        observed_accounts = failed_accounts.union(set(account_reports.keys()))
+        recorded_accounts = set(get_resources_from_redis(redis_client, 'accounts'))
 
-    stale_buckets = recorded_buckets.difference(observed_buckets)
-    _log.info('Found %s stale buckets' % len(stale_buckets))
-    for chunk in chunks(stale_buckets, ACCOUNT_UPDATE_CHUNKSIZE):
-        pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
-        for bucket in chunk:
-            update_redis(pipeline, 'buckets', bucket, 0, 0)
-            log_report('buckets', bucket, 0, 0)
-        pipeline.execute()
+        if options.account:
+            stale_accounts = { a for a in options.account if a not in observed_accounts }
+        else:
+            # Stale accounts and buckets are ones that do not appear in the listing, but have recorded values
+            stale_accounts = recorded_accounts.difference(observed_accounts)
+
+        _log.info('Found %s stale accounts' % len(stale_accounts))
+        if options.dry_run:
+            _log.info("DryRun: not updating stale accounts")
+        else:
+            for chunk in chunks(stale_accounts, ACCOUNT_UPDATE_CHUNKSIZE):
+                pipeline = redis_client.pipeline(transaction=False) # No transaction to reduce redis load
+                for account in chunk:
+                    update_redis(pipeline, 'accounts', account, 0, 0)
+                    log_report('accounts', account, 0, 0)
+                pipeline.execute()

lib/schema.js

@@ -52,6 +52,9 @@ const keys = {
     getObjectRetention: prefix => `${prefix}GetObjectRetention`,
     putObjectLegalHold: prefix => `${prefix}PutObjectLegalHold`,
     getObjectLegalHold: prefix => `${prefix}GetObjectLegalHold`,
+    replicateObject: prefix => `${prefix}ReplicateObject`,
+    replicateTags: prefix => `${prefix}ReplicateTags`,
+    replicateDelete: prefix => `${prefix}ReplicateDelete`,
     incomingBytes: prefix => `${prefix}incomingBytes`,
     outgoingBytes: prefix => `${prefix}outgoingBytes`,
 };
@@ -65,10 +68,10 @@ const keys = {
 */
 function getSchemaPrefix(params, timestamp) {
     const {
-        bucket, accountId, userId, level, service,
+        bucket, accountId, userId, level, service, location,
     } = params;
     // `service` property must remain last because other objects also include it
-    const id = bucket || accountId || userId || service;
+    const id = bucket || accountId || userId || location || service;
     const prefix = timestamp ? `${service}:${level}:${timestamp}:${id}:`
         : `${service}:${level}:${id}:`;
     return prefix;
@@ -83,9 +86,13 @@ function getSchemaPrefix(params, timestamp) {
 */
 function generateKey(params, metric, timestamp) {
     const prefix = getSchemaPrefix(params, timestamp);
+    if (params.location) {
+        return `${prefix}locationStorage`;
+    }
     return keys[metric](prefix);
 }
 
+
 /**
 * Returns a list of the counters for a metric type
 * @param {object} params - object with metric type and id as a property

lib/server.js

@@ -7,7 +7,6 @@ const { Clustering, errors, ipCheck } = require('arsenal');
 const arsenalHttps = require('arsenal').https;
 const { Logger } = require('werelogs');
 
-const config = require('./Config');
 const routes = require('../router/routes');
 const Route = require('../router/Route');
 const Router = require('../router/Router');
@@ -28,7 +27,12 @@ class UtapiServer {
     constructor(worker, port, datastore, logger, config) {
         this.worker = worker;
         this.port = port;
-        this.router = new Router(config);
+        this.vault = config.vaultclient;
+        if (!this.vault) {
+            const Vault = require('./Vault');
+            this.vault = new Vault(config);
+        }
+        this.router = new Router(config, this.vault);
         this.logger = logger;
         this.datastore = datastore;
         this.server = null;
@@ -71,6 +75,7 @@ class UtapiServer {
         req.socket.setNoDelay();
         const { query, path, pathname } = url.parse(req.url, true);
         const utapiRequest = new UtapiRequest()
+            .setVault(this.vault)
             .setRequest(req)
             .setLog(this.logger.newRequestLogger())
             .setResponse(res)
@@ -214,8 +219,7 @@ class UtapiServer {
 * @property {object} params.log - logger configuration
 * @return {undefined}
 */
-function spawn(params) {
-    Object.assign(config, params);
+function spawn(config) {
     const {
         workers, redis, log, port,
     } = config;

libV2/cache/client.js

@@ -23,7 +23,7 @@ class CacheClient {
 
     async pushMetric(metric) {
         const shard = shardFromTimestamp(metric.timestamp);
-        if (!this._cacheBackend.addToShard(shard, metric)) {
+        if (!(await this._cacheBackend.addToShard(shard, metric))) {
             return false;
         }
         await this._counterBackend.updateCounters(metric);

libV2/client/index.js

@@ -8,10 +8,16 @@ const needle = require('needle');
 const levelup = require('levelup');
 const memdown = require('memdown');
 const encode = require('encoding-down');
-const { UtapiMetric } = require('../models');
-const { LoggerContext, asyncOrCallback } = require('../utils');
 /* eslint-enable import/no-extraneous-dependencies */
 
+const { UtapiMetric } = require('../models');
+const {
+    LoggerContext,
+    logEventFilter,
+    asyncOrCallback,
+    buildFilterChain,
+} = require('../utils');
+
 const moduleLogger = new LoggerContext({
     module: 'client',
 });
@@ -70,13 +76,24 @@ class UtapiClient {
     constructor(config) {
         this._host = (config && config.host) || 'localhost';
         this._port = (config && config.port) || '8100';
+        this._tls = (config && config.tls) || {};
+        this._transport = (config && config.tls) ? 'https' : 'http';
         this._logger = (config && config.logger) || moduleLogger;
         this._maxCachedMetrics = (config && config.maxCachedMetrics) || 200000; // roughly 100MB
         this._numCachedMetrics = 0;
-        this._retryCache = levelup(encode(memdown(), { valueEncoding: 'json' }));
+        this._disableRetryCache = config && config.disableRetryCache;
+        this._retryCache = this._disableRetryCache
+            ? null
+            : levelup(encode(memdown(), { valueEncoding: 'json' }));
         this._drainTimer = null;
         this._drainCanSchedule = true;
         this._drainDelay = (config && config.drainDelay) || 30000;
+        this._suppressedEventFields = (config && config.suppressedEventFields) || null;
+        const eventFilters = (config && config.filter) || {};
+        this._shouldPushMetric = buildFilterChain(eventFilters);
+        if (Object.keys(eventFilters).length !== 0) {
+            logEventFilter((...args) => moduleLogger.info(...args), 'utapi event filter enabled', eventFilters);
+        }
     }
 
     async join() {
@@ -87,9 +104,9 @@ class UtapiClient {
     async _pushToUtapi(metrics) {
         const resp = await needle(
             'post',
-            `http://${this._host}:${this._port}/v2/ingest`,
+            `${this._transport}://${this._host}:${this._port}/v2/ingest`,
             metrics.map(metric => metric.getValue()),
-            { json: true },
+            { json: true, ...this._tls },
         );
         if (resp.statusCode !== 200) {
             throw Error('failed to push metric, server returned non 200 status code',
@@ -151,7 +168,8 @@ class UtapiClient {
         try {
             const resp = await needle(
                 'get',
-                `http://${this._host}:${this._port}/_/healthcheck`,
+                `${this._transport}://${this._host}:${this._port}/_/healthcheck`,
+                this._tls,
             );
             return resp.statusCode === 200;
         } catch (error) {
@@ -228,10 +246,15 @@ class UtapiClient {
     }
 
     async _pushMetric(data) {
-        const metric = data instanceof UtapiMetric
+        let metric = data instanceof UtapiMetric
             ? data
             : new UtapiMetric(data);
 
+        // If this event has been filtered then exit early
+        if (!this._shouldPushMetric(metric)) {
+            return;
+        }
+
         // Assign a uuid if one isn't passed
         if (!metric.uuid) {
             metric.uuid = uuid.v4();
@@ -242,12 +265,26 @@ class UtapiClient {
             metric.timestamp = new Date().getTime();
         }
 
+        if (this._suppressedEventFields !== null) {
+            const filteredData = Object.entries(metric.getValue())
+                .filter(([key]) => !this._suppressedEventFields.includes(key))
+                .reduce((obj, [key, value]) => {
+                    obj[key] = value;
+                    return obj;
+                }, {});
+            metric = new UtapiMetric(filteredData);
+        }
+
         try {
             await this._pushToUtapi([metric]);
         } catch (error) {
-            this._logger.error('unable to push metric, adding to retry cache', { error });
-            if (!await this._addToRetryCache(metric)) {
-                throw new Error('unable to store metric');
+            if (!this._disableRetryCache) {
+                this._logger.error('unable to push metric, adding to retry cache', { error });
+                if (!await this._addToRetryCache(metric)) {
+                    throw new Error('unable to store metric');
+                }
+            } else {
+                this._logger.debug('unable to push metric. retry cache disabled, not retrying ingestion.', { error });
             }
         }
     }
@@ -275,7 +312,9 @@ class UtapiClient {
         return asyncOrCallback(async () => {
             const resp = await needle(
                 'get',
-                `http://${this._host}:${this._port}/v2/storage/${level}/${resource}`,
+                `${this._transport}://${this._host}:${this._port}/v2/storage/${level}/${resource}`,
+                this._tls,
+
             );
 
             if (resp.statusCode !== 200) {

libV2/config/defaults.json

@@ -15,15 +15,14 @@
     },
     "warp10": {
         "host": "127.0.0.1",
-        "port": 4802
+        "port": 4802,
+        "nodeId": "single_node",
+        "requestTimeout": 60000,
+        "connectTimeout": 60000
     },
     "healthChecks": {
         "allowFrom": ["127.0.0.1/8", "::1"]
     },
-    "vaultd": {
-        "host": "127.0.0.1",
-        "port": 8500
-    },
     "cacheBackend": "memory",
     "development": false,
     "nodeId": "single_node",
@@ -34,9 +33,32 @@
     "snapshotSchedule": "5 0 * * * *",
     "repairSchedule": "0 */5 * * * *",
     "reindexSchedule": "0 0 0 * * Sun",
+    "diskUsageSchedule": "0 */15 * * * *",
     "bucketd": [ "localhost:9000" ],
     "reindex": {
         "enabled": true,
         "schedule": "0 0 0 * * 6"
+    },
+    "diskUsage": {
+        "retentionDays": 45,
+        "expirationEnabled": false
+    },
+    "serviceUser": {
+        "arn": "arn:aws:iam::000000000000:user/scality-internal/service-utapi-user",
+        "enabled": false
+    },
+    "filter": {
+        "allow": {},
+        "deny": {}
+    },
+    "metrics" : {
+        "enabled": false,
+        "host": "localhost",
+        "ingestPort": 10902,
+        "checkpointPort": 10903,
+        "snapshotPort": 10904,
+        "diskUsagePort": 10905,
+        "reindexPort": 10906,
+        "repairPort": 10907
     }
 }

libV2/config/index.js

@@ -2,27 +2,47 @@ const fs = require('fs');
 const path = require('path');
 const Joi = require('@hapi/joi');
 const assert = require('assert');
+const defaults = require('./defaults.json');
+const werelogs = require('werelogs');
 
-const { truthy, envNamespace } = require('../constants');
+const {
+    truthy, envNamespace, allowedFilterFields, allowedFilterStates,
+} = require('../constants');
 const configSchema = require('./schema');
+// We need to require the specific file rather than the parent module to avoid a circular require
+const { parseDiskSizeSpec } = require('../utils/disk');
+
+function _splitTrim(char, text) {
+    return text.split(char).map(v => v.trim());
+}
 
 function _splitServer(text) {
     assert.notStrictEqual(text.indexOf(':'), -1);
-    const [host, port] = text.split(':').map(v => v.trim());
+    const [host, port] = _splitTrim(':', text);
     return {
         host,
         port: Number.parseInt(port, 10),
     };
 }
 
+function _splitNode(text) {
+    assert.notStrictEqual(text.indexOf('='), -1);
+    const [nodeId, hostname] = _splitTrim('=', text);
+    return {
+        nodeId,
+        ..._splitServer(hostname),
+    };
+}
+
 const _typeCasts = {
     bool: val => truthy.has(val.toLowerCase()),
     int: val => parseInt(val, 10),
-    list: val => val.split(',').map(v => v.trim()),
-    serverList: val => val.split(',').map(v => v.trim()).map(_splitServer),
+    list: val => _splitTrim(',', val),
+    serverList: val => _splitTrim(',', val).map(_splitServer),
+    nodeList: val => _splitTrim(',', val).map(_splitNode),
+    diskSize: parseDiskSizeSpec,
 };
 
-
 function _definedInEnv(key) {
     return process.env[`${envNamespace}_${key}`] !== undefined;
 }
@@ -53,7 +73,6 @@ class Config {
     constructor(overrides) {
         this._basePath = path.join(__dirname, '../../');
         this._configPath = _loadFromEnv('CONFIG_FILE', defaultConfigPath);
-        this._defaultsPath = path.join(__dirname, 'defaults.json');
 
         this.host = undefined;
         this.port = undefined;
@@ -71,6 +90,11 @@ class Config {
             parsedConfig = this._recursiveUpdate(parsedConfig, overrides);
         }
         Object.assign(this, parsedConfig);
+
+        werelogs.configure({
+            level: Config.logging.level,
+            dump: Config.logging.dumpLevel,
+        });
     }
 
     static _readFile(path, encoding = 'utf-8') {
@@ -95,7 +119,7 @@ class Config {
     }
 
     _loadDefaults() {
-        return Config._readJSON(this._defaultsPath);
+        return defaults;
     }
 
     _loadUserConfig() {
@@ -159,38 +183,37 @@ class Config {
         return this._recursiveUpdateObject(defaultConf, userConf);
     }
 
-    static _parseRedisConfig(config) {
-        const redisConf = {};
-        if (config.sentinels || _definedInEnv('REDIS_SENTINELS')) {
-            redisConf.name = _loadFromEnv('REDIS_NAME', config.name);
-            const sentinels = _loadFromEnv(
-                'REDIS_SENTINELS',
+    static _parseRedisConfig(prefix, config) {
+        const redisConf = {
+            retry: config.retry,
+        };
+        if (config.sentinels || _definedInEnv(`${prefix}_SENTINELS`)) {
+            redisConf.name = _loadFromEnv(`${prefix}_NAME`, config.name);
+            redisConf.sentinels = _loadFromEnv(
+                `${prefix}_SENTINELS`,
                 config.sentinels,
-                _typeCasts.list,
+                _typeCasts.serverList,
             );
-            redisConf.sentinels = sentinels.map(v => {
-                if (typeof v === 'string') {
-                    const [host, port] = v.split(':');
-                    return { host, port: Number.parseInt(port, 10) };
-                }
-                return v;
-            });
             redisConf.sentinelPassword = _loadFromEnv(
-                'REDIS_SENTINEL_PASSWORD',
+                `${prefix}_SENTINEL_PASSWORD`,
                 config.sentinelPassword,
             );
+            redisConf.password = _loadFromEnv(
+                `${prefix}_PASSWORD`,
+                config.password,
+            );
         } else {
             redisConf.host = _loadFromEnv(
-                'REDIS_HOST',
+                `${prefix}_HOST`,
                 config.host,
             );
             redisConf.port = _loadFromEnv(
-                'REDIS_PORT',
+                `${prefix}_PORT`,
                 config.port,
                 _typeCasts.int,
             );
             redisConf.password = _loadFromEnv(
-                'REDIS_PASSWORD',
+                `${prefix}_PASSWORD`,
                 config.password,
             );
         }
@@ -216,6 +239,28 @@ class Config {
         return certs;
     }
 
+    static _parseResourceFilters(config) {
+        const resourceFilters = {};
+
+        allowedFilterFields.forEach(
+            field => allowedFilterStates.forEach(
+                state => {
+                    const configResources = (config[state] && config[state][field]) || null;
+                    const envVar = `FILTER_${field.toUpperCase()}_${state.toUpperCase()}`;
+                    const resources = _loadFromEnv(envVar, configResources, _typeCasts.list);
+                    if (resources) {
+                        if (resourceFilters[field]) {
+                            throw new Error('You can not define both an allow and a deny list for an event field.');
+                        }
+                        resourceFilters[field] = { [state]: new Set(resources) };
+                    }
+                },
+            ),
+        );
+
+        return resourceFilters;
+    }
+
     _parseConfig(config) {
         const parsedConfig = {};
 
@@ -246,25 +291,30 @@ class Config {
             throw new Error('bad config: both certFilePaths.key and certFilePaths.cert must be defined');
         }
 
-        parsedConfig.redis = Config._parseRedisConfig(config.redis);
+        parsedConfig.redis = Config._parseRedisConfig('REDIS', config.redis);
 
-        parsedConfig.cache = Config._parseRedisConfig(config.localCache);
+        parsedConfig.cache = Config._parseRedisConfig('REDIS_CACHE', config.localCache);
         parsedConfig.cache.backend = _loadFromEnv('CACHE_BACKEND', config.cacheBackend);
 
         const warp10Conf = {
             readToken: _loadFromEnv('WARP10_READ_TOKEN', config.warp10.readToken),
             writeToken: _loadFromEnv('WARP10_WRITE_TOKEN', config.warp10.writeToken),
+            requestTimeout: _loadFromEnv('WARP10_REQUEST_TIMEOUT', config.warp10.requestTimeout, _typeCasts.int),
+            connectTimeout: _loadFromEnv('WARP10_CONNECT_TIMEOUT', config.warp10.connectTimeout, _typeCasts.int),
         };
 
-        parsedConfig.warp10 = warp10Conf;
-
         if (Array.isArray(config.warp10.hosts) || _definedInEnv('WARP10_HOSTS')) {
-            warp10Conf.hosts = _loadFromEnv('WARP10_HOSTS', config.warp10.hosts, _typeCasts.serverList);
+            warp10Conf.hosts = _loadFromEnv('WARP10_HOSTS', config.warp10.hosts, _typeCasts.nodeList);
         } else {
-            warp10Conf.host = _loadFromEnv('WARP10_HOST', config.warp10.host);
-            warp10Conf.port = _loadFromEnv('WARP10_PORT', config.warp10.port, _typeCasts.int);
+            warp10Conf.hosts = [{
+                host: _loadFromEnv('WARP10_HOST', config.warp10.host),
+                port: _loadFromEnv('WARP10_PORT', config.warp10.port, _typeCasts.int),
+                nodeId: _loadFromEnv('WARP10_NODE_ID', config.warp10.nodeId),
+            }];
         }
 
+        parsedConfig.warp10 = warp10Conf;
+
         parsedConfig.logging = {
             level: parsedConfig.development
                 ? 'debug'
@@ -280,6 +330,7 @@ class Config {
         parsedConfig.snapshotSchedule = _loadFromEnv('SNAPSHOT_SCHEDULE', config.snapshotSchedule);
         parsedConfig.repairSchedule = _loadFromEnv('REPAIR_SCHEDULE', config.repairSchedule);
         parsedConfig.reindexSchedule = _loadFromEnv('REINDEX_SCHEDULE', config.reindexSchedule);
+        parsedConfig.diskUsageSchedule = _loadFromEnv('DISK_USAGE_SCHEDULE', config.diskUsageSchedule);
 
         parsedConfig.ingestionLagSeconds = _loadFromEnv(
             'INGESTION_LAG_SECONDS',
@@ -292,6 +343,34 @@ class Config {
             _typeCasts.int,
         );
 
+        const diskUsage = {
+            path: _loadFromEnv('DISK_USAGE_PATH', (config.diskUsage || {}).path),
+            hardLimit: _loadFromEnv('DISK_USAGE_HARD_LIMIT', (config.diskUsage || {}).hardLimit),
+            retentionDays: _loadFromEnv(
+                'METRIC_RETENTION_PERIOD',
+                (config.diskUsage || {}).retentionDays, _typeCasts.int,
+            ),
+            expirationEnabled: _loadFromEnv(
+                'METRIC_EXPIRATION_ENABLED',
+                (config.diskUsage || {}).expirationEnabled, _typeCasts.bool,
+            ),
+        };
+
+        if (diskUsage.hardLimit !== undefined) {
+            diskUsage.hardLimit = parseDiskSizeSpec(diskUsage.hardLimit);
+        }
+
+        if (!diskUsage.path && diskUsage.hardLimit !== undefined) {
+            throw Error('You must specify diskUsage.path to monitor for disk usage');
+        } else if (diskUsage.path && diskUsage.hardLimit === undefined) {
+            throw Error('diskUsage.hardLimit must be specified');
+        } else if (diskUsage.expirationEnabled && diskUsage.retentionDays === undefined) {
+            throw Error('diskUsage.retentionDays must be specified');
+        }
+
+        diskUsage.enabled = diskUsage.path !== undefined;
+        parsedConfig.diskUsage = diskUsage;
+
         parsedConfig.vaultd = {
             host: _loadFromEnv('VAULT_HOST', config.vaultd.host),
             port: _loadFromEnv('VAULT_PORT', config.vaultd.port),
@@ -299,6 +378,24 @@ class Config {
 
         parsedConfig.bucketd = _loadFromEnv('BUCKETD_BOOTSTRAP', config.bucketd, _typeCasts.serverList);
 
+        parsedConfig.serviceUser = {
+            arn: _loadFromEnv('SERVICE_USER_ARN', config.serviceUser.arn),
+            enabled: _loadFromEnv('SERVICE_USER_ENABLED', config.serviceUser.enabled, _typeCasts.bool),
+        };
+
+        parsedConfig.filter = Config._parseResourceFilters(config.filter);
+
+        parsedConfig.metrics = {
+            enabled: _loadFromEnv('METRICS_ENABLED', config.metrics.enabled, _typeCasts.bool),
+            host: _loadFromEnv('METRICS_HOST', config.metrics.host),
+            ingestPort: _loadFromEnv('METRICS_PORT_INGEST', config.metrics.ingestPort, _typeCasts.int),
+            checkpointPort: _loadFromEnv('METRICS_PORT_CHECKPOINT', config.metrics.checkpointPort, _typeCasts.int),
+            snapshotPort: _loadFromEnv('METRICS_PORT_SNAPSHOT', config.metrics.snapshotPort, _typeCasts.int),
+            diskUsagePort: _loadFromEnv('METRICS_PORT_DISK_USAGE', config.metrics.diskUsagePort, _typeCasts.int),
+            reindexPort: _loadFromEnv('METRICS_PORT_REINDEX', config.metrics.reindexPort, _typeCasts.int),
+            repairPort: _loadFromEnv('METRICS_PORT_REPAIR', config.metrics.repairPort, _typeCasts.int),
+        };
+
         return parsedConfig;
     }
 

libV2/config/schema.js

@@ -1,9 +1,23 @@
 const Joi = require('@hapi/joi');
+const { allowedFilterFields, allowedFilterStates } = require('../constants');
+
+const backoffSchema = Joi.object({
+    min: Joi.number(),
+    max: Joi.number(),
+    deadline: Joi.number(),
+    jitter: Joi.number(),
+    factor: Joi.number(),
+});
+
+const redisRetrySchema = Joi.object({
+    connectBackoff: backoffSchema,
+});
 
 const redisServerSchema = Joi.object({
     host: Joi.string(),
     port: Joi.number(),
     password: Joi.string().allow(''),
+    retry: redisRetrySchema,
 });
 
 const redisSentinelSchema = Joi.object({
@@ -14,6 +28,7 @@ const redisSentinelSchema = Joi.object({
     })),
     password: Joi.string().default('').allow(''),
     sentinelPassword: Joi.string().default('').allow(''),
+    retry: redisRetrySchema,
 });
 
 const warp10SingleHost = Joi.object({
@@ -27,6 +42,7 @@ const warp10MultiHost = Joi.object({
     hosts: Joi.array().items(Joi.object({
         host: Joi.alternatives(Joi.string().hostname(), Joi.string().ip()),
         port: Joi.number().port(),
+        nodeId: Joi.string(),
     })),
     readToken: Joi.string(),
     writeToken: Joi.string(),
@@ -77,7 +93,38 @@ const schema = Joi.object({
     snapshotSchedule: Joi.string(),
     repairSchedule: Joi.string(),
     reindexSchedule: Joi.string(),
+    diskUsageSchedule: Joi.string(),
+    diskUsage: Joi.object({
+        path: Joi.string(),
+        retentionDays: Joi.number().greater(0),
+        expirationEnabled: Joi.boolean(),
+        hardLimit: Joi.string(),
+    }),
+    serviceUser: Joi.object({
+        arn: Joi.string(),
+        enabled: Joi.boolean(),
+    }),
+    filter: Joi.object(allowedFilterStates.reduce(
+        (filterObj, state) => {
+            filterObj[state] = allowedFilterFields.reduce(
+                (stateObj, field) => {
+                    stateObj[field] = Joi.array().items(Joi.string());
+                    return stateObj;
+                }, {},
+            );
+            return filterObj;
+        }, {},
+    )),
+    metrics: {
+        enabled: Joi.boolean(),
+        host: Joi.string(),
+        ingestPort: Joi.number().port(),
+        checkpointPort: Joi.number().port(),
+        snapshotPort: Joi.number().port(),
+        diskUsagePort: Joi.number().port(),
+        reindexPort: Joi.number().port(),
+        repairPort: Joi.number().port(),
+    },
 });
 
 module.exports = schema;
-

libV2/constants.js

@@ -19,17 +19,23 @@ const constants = {
         'createBucket',
         'deleteBucket',
         'deleteBucketCors',
+        'deleteBucketEncryption',
+        'deleteBucketLifecycle',
         'deleteBucketReplication',
+        'deleteBucketTagging',
         'deleteBucketWebsite',
         'deleteObject',
         'deleteObjectTagging',
         'getBucketAcl',
         'getBucketCors',
+        'getBucketEncryption',
+        'getBucketLifecycle',
         'getBucketLocation',
         'getBucketNotification',
         'getBucketObjectLock',
         'getBucketReplication',
         'getBucketVersioning',
+        'getBucketTagging',
         'getBucketWebsite',
         'getObject',
         'getObjectAcl',
@@ -45,18 +51,23 @@ const constants = {
         'multiObjectDelete',
         'putBucketAcl',
         'putBucketCors',
+        'putBucketEncryption',
+        'putBucketLifecycle',
         'putBucketNotification',
         'putBucketObjectLock',
         'putBucketReplication',
         'putBucketVersioning',
+        'putBucketTagging',
         'putBucketWebsite',
-        'putData',
         'putDeleteMarkerObject',
         'putObject',
         'putObjectAcl',
         'putObjectLegalHold',
         'putObjectRetention',
         'putObjectTagging',
+        'replicateDelete',
+        'replicateObject',
+        'replicateTags',
         'uploadPart',
         'uploadPartCopy',
     ],
@@ -96,6 +107,21 @@ const constants = {
     counterBaseValueExpiration: 86400, // 24hrs
     keyVersionSplitter: String.fromCharCode(0),
     migrationChunksize: 500,
+    migrationOpTranslationMap: {
+        listBucketMultipartUploads: 'listMultipartUploads',
+    },
+    ingestionOpTranslationMap: {
+        putDeleteMarkerObject: 'deleteObject',
+    },
+    expirationChunkDuration: 900000000, // 15 minutes in microseconds
+    allowedFilterFields: [
+        'operationId',
+        'location',
+        'account',
+        'user',
+        'bucket',
+    ],
+    allowedFilterStates: ['allow', 'deny'],
 };
 
 constants.operationToResponse = constants.operations

libV2/errors/errors.json

@@ -1,7 +1,7 @@
 {
     "AccessDenied": {
         "code": 403,
-        "description": "Access denied"
+        "description": "Access Denied"
     },
     "InternalError": {
         "code": 500,

libV2/metadata/client.js

@@ -1,5 +1,5 @@
-const BucketClientInterface = require('arsenal/lib/storage/metadata/bucketclient/BucketClientInterface');
 const bucketclient = require('bucketclient');
+const { BucketClientInterface } = require('arsenal').storage.metadata.bucketclient;
 
 const config = require('../config');
 const { LoggerContext } = require('../utils');
@@ -10,7 +10,7 @@ const moduleLogger = new LoggerContext({
 
 const params = {
     bucketdBootstrap: config.bucketd,
-    https: config.https,
+    https: config.tls,
 };
 
 module.exports = new BucketClientInterface(params, bucketclient, moduleLogger);

libV2/metadata/index.js

@@ -1,16 +1,25 @@
 /* eslint-disable no-restricted-syntax */
-const { usersBucket, splitter: mdKeySplitter, mpuBucketPrefix } = require('arsenal').constants;
+const arsenal = require('arsenal');
+const async = require('async');
 const metadata = require('./client');
 const { LoggerContext, logger } = require('../utils');
 const { keyVersionSplitter } = require('../constants');
 
+const { usersBucket, splitter: mdKeySplitter, mpuBucketPrefix } = arsenal.constants;
+const { BucketInfo } = arsenal.models;
+
 const moduleLogger = new LoggerContext({
     module: 'metadata.client',
 });
 
+const ebConfig = {
+    times: 10,
+    interval: retryCount => 50 * (2 ** retryCount),
+};
+
 const PAGE_SIZE = 1000;
 
-function _listingWrapper(bucket, params) {
+async function _listingWrapper(bucket, params) {
     return new Promise(
         (resolve, reject) => metadata.listObject(
             bucket,
@@ -37,7 +46,7 @@ function _listObject(bucket, prefix, hydrateFunc) {
 
                 try {
                     // eslint-disable-next-line no-await-in-loop
-                    res = await _listingWrapper(bucket, { ...listingParams, gt });
+                    res = await async.retryable(ebConfig, _listingWrapper)(bucket, { ...listingParams, gt });
                 } catch (error) {
                     moduleLogger.error('Error during listing', { error });
                     throw error;
@@ -99,7 +108,7 @@ function bucketExists(bucket) {
         bucket,
         logger.newRequestLogger(),
         err => {
-            if (err && !err.NoSuchBucket) {
+            if (err && (!err.is || !err.is.NoSuchBucket)) {
                 reject(err);
                 return;
             }
@@ -108,9 +117,25 @@ function bucketExists(bucket) {
     ));
 }
 
+function getBucket(bucket) {
+    return new Promise((resolve, reject) => {
+        metadata.getBucketAttributes(
+            bucket,
+            logger.newRequestLogger(), (err, data) => {
+                if (err) {
+                    reject(err);
+                    return;
+                }
+                resolve(BucketInfo.fromObj(data));
+            },
+        );
+    });
+}
+
 module.exports = {
     listBuckets,
     listObjects,
     listMPUs,
     bucketExists,
+    getBucket,
 };

libV2/models/RequestContext.js

@@ -3,6 +3,7 @@ const Joi = require('@hapi/joi');
 const { buildModel } = require('./Base');
 const { apiOperations } = require('../server/spec');
 const ResponseContainer = require('./ResponseContainer');
+const { httpRequestDurationSeconds } = require('../server/metrics');
 
 const apiTags = Object.keys(apiOperations);
 const apiOperationIds = Object.values(apiOperations)
@@ -21,6 +22,7 @@ const contextSchema = {
     logger: Joi.any(),
     request: Joi.any(),
     results: Joi.any(),
+    requestTimer: Joi.any(),
 };
 
 const RequestContextModel = buildModel('RequestContext', contextSchema);
@@ -34,6 +36,10 @@ class RequestContext extends RequestContextModel {
         const tag = request.swagger.operation['x-router-controller'];
         const { operationId } = request.swagger.operation;
 
+        const requestTimer = tag !== 'internal'
+            ? httpRequestDurationSeconds.startTimer({ action: operationId })
+            : null;
+
         request.logger.logger.addDefaultFields({
             tag,
             operationId,
@@ -50,6 +56,7 @@ class RequestContext extends RequestContextModel {
             encrypted,
             results: new ResponseContainer(),
             logger: request.logger,
+            requestTimer,
         });
     }
 

libV2/redis.js

@@ -2,9 +2,12 @@ const EventEmitter = require('events');
 const { callbackify, promisify } = require('util');
 const IORedis = require('ioredis');
 const { jsutil } = require('arsenal');
+const BackOff = require('backo');
+const { whilst } = require('async');
 
 const errors = require('./errors');
-const { LoggerContext, asyncOrCallback } = require('./utils');
+const { LoggerContext } = require('./utils/log');
+const { asyncOrCallback } = require('./utils/func');
 
 const moduleLogger = new LoggerContext({
     module: 'redis',
@@ -51,7 +54,10 @@ class RedisClient extends EventEmitter {
                 Object.values(this._inFlightTimeouts)
                     .forEach(clearTimeout);
             }
-            await this._redis.quit();
+            if (this._redis !== null) {
+                await this._redis.quit();
+                this._redis = null;
+            }
         }, callback);
     }
 
@@ -65,6 +71,7 @@ class RedisClient extends EventEmitter {
             this._redis.off('connect', this._onConnect);
             this._redis.off('ready', this._onReady);
             this._redis.off('error', this._onError);
+            this._redis.disconnect();
         }
         this._isConnected = false;
         this._isReady = false;
@@ -99,53 +106,96 @@ class RedisClient extends EventEmitter {
     }
 
     _onError(error) {
+        this._isReady = false;
         moduleLogger.error('error connecting to redis', { error });
-        this.emit('error', error);
+        if (this.listenerCount('error') > 0) {
+            this.emit('error', error);
+        }
     }
 
-
     _createCommandTimeout() {
         let timer;
+        let onTimeout;
+
         const cancelTimeout = jsutil.once(() => {
             clearTimeout(timer);
+            this.off('timeout', onTimeout);
             this._inFlightTimeouts.delete(timer);
         });
 
         const timeout = new Promise((_, reject) => {
-            timer = setTimeout(
-                () => {
-                    this.emit('timeout');
-                    this._initClient();
-                },
-                COMMAND_TIMEOUT,
-            );
-
+            timer = setTimeout(this.emit.bind(this, 'timeout'), COMMAND_TIMEOUT);
             this._inFlightTimeouts.add(timer);
-            this.once('timeout', () => {
+            onTimeout = () => {
                 moduleLogger.warn('redis command timed out');
                 cancelTimeout();
+                this._initClient();
                 reject(errors.OperationTimedOut);
-            });
+            };
+            this.once('timeout', onTimeout);
         });
 
         return { timeout, cancelTimeout };
     }
 
     async _call(asyncFunc) {
-        const funcPromise = asyncFunc(this._redis);
-        if (!this._useTimeouts) {
-            // If timeouts are disabled simply return the Promise
-            return funcPromise;
-        }
+        const start = Date.now();
+        const { connectBackoff } = this._redisOptions.retry || {};
+        const backoff = new BackOff(connectBackoff);
+        const timeoutMs = (connectBackoff || {}).deadline || 2000;
+        let retried = false;
 
-        const { timeout, cancelTimeout } = this._createCommandTimeout();
+        return new Promise((resolve, reject) => {
+            whilst(
+                next => { // WARNING: test is asynchronous in `async` v3
+                    if (!connectBackoff && !this.isReady) {
+                        moduleLogger.warn('redis not ready and backoff is not configured');
+                    }
+                    process.nextTick(next, null, !!connectBackoff && !this.isReady);
+                },
+                next => {
+                    retried = true;
 
-        try {
-            // timeout always rejects so we can just return
-            return await Promise.race([funcPromise, timeout]);
-        } finally {
-            cancelTimeout();
-        }
+                    if ((Date.now() - start) > timeoutMs) {
+                        moduleLogger.error('redis still not ready after max wait, giving up', { timeoutMs });
+                        return next(errors.InternalError.customizeDescription(
+                            'redis client is not ready',
+                        ));
+                    }
+
+                    const backoffDurationMs = backoff.duration();
+                    moduleLogger.error('redis not ready, retrying', { backoffDurationMs });
+
+                    return setTimeout(next, backoffDurationMs);
+                },
+                err => {
+                    if (err) {
+                        return reject(err);
+                    }
+
+                    if (retried) {
+                        moduleLogger.info('redis connection recovered', {
+                            recoveryOverheadMs: Date.now() - start,
+                        });
+                    }
+
+                    const funcPromise = asyncFunc(this._redis);
+                    if (!this._useTimeouts) {
+                        // If timeouts are disabled simply return the Promise
+                        return resolve(funcPromise);
+                    }
+
+                    const { timeout, cancelTimeout } = this._createCommandTimeout();
+
+                    try {
+                        // timeout always rejects so we can just return
+                        return resolve(Promise.race([funcPromise, timeout]));
+                    } finally {
+                        cancelTimeout();
+                    }
+                },
+            );
+        });
     }
 
     call(func, callback) {

libV2/server/API/internal/prometheusMetrics.js

@@ -0,0 +1,14 @@
+const { collectDefaultMetrics, register } = require('prom-client');
+
+collectDefaultMetrics({
+    timeout: 10000,
+    gcDurationBuckets: [0.001, 0.01, 0.1, 1, 2, 5],
+});
+
+async function prometheusMetrics(ctx) {
+    // eslint-disable-next-line no-param-reassign
+    ctx.results.statusCode = 200;
+    ctx.results.body = await register.metrics();
+}
+
+module.exports = prometheusMetrics;

libV2/server/API/metrics/getStorage.js

@@ -1,9 +1,8 @@
 const errors = require('../../../errors');
 const { serviceToWarp10Label } = require('../../../constants');
-const { client: warp10 } = require('../../../warp10');
+const { clients: warp10Clients } = require('../../../warp10');
 const { client: cache } = require('../../../cache');
-const { now } = require('../../../utils');
-const config = require('../../../config');
+const { now, iterIfError } = require('../../../utils');
 
 /**
  *
@@ -30,15 +29,18 @@ async function getStorage(ctx, params) {
     } else {
         const labelName = serviceToWarp10Label[params.level];
         const labels = { [labelName]: resource };
-        const options = {
-            params: {
-                end: now(),
-                labels,
-                node: config.nodeId,
-            },
-            macro: 'utapi/getMetricsAt',
-        };
-        const res = await warp10.exec(options);
+
+        const res = await iterIfError(warp10Clients, warp10 => {
+            const options = {
+                params: {
+                    end: now(),
+                    labels,
+                    node: warp10.nodeId,
+                },
+                macro: 'utapi/getMetricsAt',
+            };
+            return warp10.exec(options);
+        }, error => ctx.logger.error('error while fetching metrics', { error }));
 
         if (res.result.length === 0) {
             ctx.logger.error('unable to retrieve metrics', { level, resource });
@@ -52,7 +54,7 @@ async function getStorage(ctx, params) {
 
     ctx.results.statusCode = 200;
     ctx.results.body = {
-        storageUtilized,
+        storageUtilized: Math.max(storageUtilized, 0),
         resource,
         level,
     };

libV2/server/API/metrics/ingestMetric.js

@@ -2,6 +2,7 @@ const errors = require('../../../errors');
 const { UtapiMetric } = require('../../../models');
 const { client: cacheClient } = require('../../../cache');
 const { convertTimestamp } = require('../../../utils');
+const { ingestionOpTranslationMap } = require('../../../constants');
 
 async function ingestMetric(ctx, params) {
     let metrics;
@@ -9,6 +10,7 @@ async function ingestMetric(ctx, params) {
         metrics = params.body.map(m => new UtapiMetric({
             ...m,
             timestamp: convertTimestamp(m.timestamp),
+            operationId: ingestionOpTranslationMap[m.operationId] || m.operationId,
         }));
     } catch (error) {
         throw errors.InvalidRequest;

libV2/server/API/metrics/listMetrics.js

@@ -1,8 +1,7 @@
 const errors = require('../../../errors');
 const { serviceToWarp10Label, operationToResponse } = require('../../../constants');
-const { convertTimestamp } = require('../../../utils');
-const { client: warp10 } = require('../../../warp10');
-const config = require('../../../config');
+const { convertTimestamp, iterIfError } = require('../../../utils');
+const { clients: warp10Clients } = require('../../../warp10');
 
 const emptyOperationsResponse = Object.values(operationToResponse)
     .reduce((prev, key) => {
@@ -17,37 +16,74 @@ const metricResponseKeys = {
     service: 'serviceName',
 };
 
+function positiveOrZero(value) {
+    return Math.max(value, 0);
+}
+
 async function listMetric(ctx, params) {
     const labelName = serviceToWarp10Label[params.level];
     const resources = params.body[params.level];
-    const [start, end] = params.body.timeRange
-        .map(convertTimestamp)
-        .map(v => v.toString());
+    let [start, end] = params.body.timeRange;
+    if (end === undefined) {
+        end = Date.now();
+    }
+
+    let results;
+
+    try {
+        // A separate request will be made to warp 10 per requested resource
+        results = await Promise.all(
+            resources.map(async ({ resource, id }) => {
+                const labels = { [labelName]: id };
+
+                const res = await iterIfError(warp10Clients, warp10 => {
+                    const options = {
+                        params: {
+                            start: convertTimestamp(start).toString(),
+                            end: convertTimestamp(end).toString(),
+                            labels,
+                            node: warp10.nodeId,
+                        },
+                        macro: 'utapi/getMetrics',
+                    };
+                    return warp10.exec(options);
+                }, error => ctx.logger.error('error during warp 10 request', {
+                    error,
+                    requestParams: {
+                        start,
+                        end,
+                        labels,
+                    },
+                }));
+
+                if (res.result.length === 0) {
+                    ctx.logger.error('unable to retrieve metrics', { resource, type: params.level });
+                    throw errors.InternalError;
+                }
+
+                const rawMetrics = JSON.parse(res.result[0]);
+
+                // Due to various error cases it is possible for metrics in utapi to go negative.
+                // As this is nonsensical to the user we replace any negative values with zero.
+                const metrics = {
+                    storageUtilized: rawMetrics.storageUtilized.map(positiveOrZero),
+                    numberOfObjects: rawMetrics.numberOfObjects.map(positiveOrZero),
+                    incomingBytes: positiveOrZero(rawMetrics.incomingBytes),
+                    outgoingBytes: positiveOrZero(rawMetrics.outgoingBytes),
+                    operations: rawMetrics.operations,
+                };
+
+                return {
+                    resource,
+                    metrics,
+                };
+            }),
+        );
+    } catch (error) {
+        ctx.logger.error('error fetching metrics from warp10', { error });
+        throw errors.InternalError;
+    }
 
-    // A separate request will be made to warp 10 per requested resource
-    const results = await Promise.all(
-        resources.map(async resource => {
-            const labels = { [labelName]: resource };
-            const options = {
-                params: {
-                    start,
-                    end,
-                    labels,
-                    node: config.nodeId,
-                },
-                macro: 'utapi/getMetrics',
-            };
-            const res = await warp10.exec(options);
-            if (res.result.length === 0) {
-                ctx.logger.error('unable to retrieve metrics', { resource, type: params.level });
-                throw errors.InternalError;
-            }
-            return {
-                resource,
-                metrics: JSON.parse(res.result[0]),
-            };
-        }),
-    );
 
     // Convert the results from warp10 into the expected response format
     const resp = results
@@ -60,7 +96,7 @@ async function listMetric(ctx, params) {
 
             const metric = {
                 ...result.metrics,
-                timeRange: params.body.timeRange,
+                timeRange: [start, end],
                 operations: {
                     ...emptyOperationsResponse,
                     ...operations,

libV2/server/index.js

@@ -28,6 +28,7 @@ class UtapiServer extends Process {
         app.use(middleware.loggerMiddleware);
         await initializeOasTools(spec, app);
         app.use(middleware.errorMiddleware);
+        app.use(middleware.httpMetricsMiddleware);
         app.use(middleware.responseLoggerMiddleware);
         return app;
     }
@@ -35,7 +36,7 @@ class UtapiServer extends Process {
     static _createHttpsAgent() {
         const conf = {
             ciphers: ciphers.ciphers,
-            dhparam,
+            dhparam: dhparam.dhparam,
             cert: config.tls.cert,
             key: config.tls.key,
             ca: config.tls.ca ? [config.tls.ca] : null,

libV2/server/metrics.js

@@ -0,0 +1,20 @@
+const promClient = require('prom-client');
+
+const httpRequestsTotal = new promClient.Counter({
+    name: 's3_utapi_http_requests_total',
+    help: 'Total number of HTTP requests',
+    labelNames: ['action', 'code'],
+});
+
+const httpRequestDurationSeconds = new promClient.Histogram({
+    name: 's3_utapi_http_request_duration_seconds',
+    help: 'Duration of HTTP requests in seconds',
+    labelNames: ['action', 'code'],
+    // buckets for response time from 0.1ms to 60s
+    buckets: [0.0001, 0.005, 0.015, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 1.0, 5.0, 15.0, 30.0, 60.0],
+});
+
+module.exports = {
+    httpRequestDurationSeconds,
+    httpRequestsTotal,
+};

libV2/server/middleware.js

@@ -5,7 +5,8 @@ const { ipCheck } = require('arsenal');
 const config = require('../config');
 const { logger, buildRequestLogger } = require('../utils');
 const errors = require('../errors');
-const { authenticateRequest, vault } = require('../vault');
+const { translateAndAuthorize } = require('../vault');
+const metricHandlers = require('./metrics');
 
 const oasOptions = {
     controllers: path.join(__dirname, './API/'),
@@ -44,41 +45,62 @@ function loggerMiddleware(req, res, next) {
     return next();
 }
 
-// next is purposely not called as all error responses are handled here
-// eslint-disable-next-line no-unused-vars
-function errorMiddleware(err, req, res, next) {
-    let code = err.code || 500;
-    let message = err.message || 'Internal Error';
-
-    // failed request validation by oas-tools
-    if (err.failedValidation) {
-        // You can't actually use destructing here
-        /* eslint-disable prefer-destructuring */
-        code = errors.InvalidRequest.code;
-        message = errors.InvalidRequest.message;
-        /* eslint-enable prefer-destructuring */
-    }
-
-    if (!err.utapiError && !config.development) {
-        // Make sure internal errors don't leak when not in development
-        message = 'Internal Error';
-    }
-
-    res.status(code).send({
-        error: {
-            code: code.toString(),
-            message,
-        },
-    });
-}
-
 function responseLoggerMiddleware(req, res, next) {
     const info = {
         httpCode: res.statusCode,
         httpMessage: res.statusMessage,
     };
     req.logger.end('finished handling request', info);
-    return next();
+    if (next !== undefined) {
+        next();
+    }
+}
+
+function httpMetricsMiddleware(request, response, next) {
+    // If the request.ctx is undefined then this is an internal oasTools request (/_/docs)
+    // No metrics should be pushed
+    if (config.metrics.enabled && request.ctx && request.ctx.tag !== 'internal') {
+        metricHandlers.httpRequestsTotal
+            .labels({
+                action: request.ctx.operationId,
+                code: response.statusCode,
+            }).inc(1);
+        request.ctx.requestTimer({ code: response.statusCode });
+    }
+
+    if (next) {
+        next();
+    }
+}
+
+// next is purposely not called as all error responses are handled here
+// eslint-disable-next-line no-unused-vars
+function errorMiddleware(err, req, res, next) {
+    let statusCode = err.code || 500;
+    let code = err.message || 'InternalError';
+    let message = err.description || 'Internal Error';
+
+    // failed request validation by oas-tools
+    if (err.failedValidation) {
+        // You can't actually use destructing here
+        /* eslint-disable prefer-destructuring */
+        statusCode = errors.InvalidRequest.code;
+        code = errors.InvalidRequest.message;
+        message = errors.InvalidRequest.description;
+        /* eslint-enable prefer-destructuring */
+    }
+
+    if (!err.utapiError && !config.development) {
+        // Make sure internal errors don't leak when not in development
+        code = 'InternalError';
+        message = 'Internal Error';
+    }
+
+    res.status(statusCode).send({
+        code,
+        message,
+    });
+    responseLoggerMiddleware(req, res, () => httpMetricsMiddleware(req, res));
 }
 
 // eslint-disable-next-line no-unused-vars
@@ -95,7 +117,7 @@ async function authV4Middleware(request, response, params) {
     switch (request.ctx.operationId) {
     case 'listMetrics':
         requestedResources = params.body[params.level];
-        action = params.Action.value;
+        action = params.Action;
         break;
 
     default:
@@ -111,9 +133,13 @@ async function authV4Middleware(request, response, params) {
     let authorizedResources;
 
     try {
-        [passed, authorizedResources] = await authenticateRequest(request, action, params.level, requestedResources);
+        [passed, authorizedResources] = await translateAndAuthorize(request, action, params.level, requestedResources);
     } catch (error) {
         request.logger.error('error during authentication', { error });
+        // rethrow any access denied errors
+        if ((error.is && error.is.AccessDenied) || (error.utapiError && error.AccessDenied)) {
+            throw error;
+        }
         throw errors.InternalError;
     }
 
@@ -122,17 +148,14 @@ async function authV4Middleware(request, response, params) {
         throw errors.AccessDenied;
     }
 
-    if (params.level === 'accounts') {
-        request.logger.debug('converting account ids to canonical ids');
-        authorizedResources = await vault.getCanonicalIds(
-            authorizedResources,
-            request.logger.logger,
-        );
-    }
-
-    // authorizedResources is only defined on non-account credentials
-    if (request.ctx.operationId === 'listMetrics' && authorizedResources !== undefined) {
+    switch (request.ctx.operationId) {
+    case 'listMetrics':
         params.body[params.level] = authorizedResources;
+        break;
+
+    default:
+        [params.resource] = authorizedResources;
+        break;
     }
 }
 
@@ -153,5 +176,6 @@ module.exports = {
         responseLoggerMiddleware,
         authV4Middleware,
         clientIpLimitMiddleware,
+        httpMetricsMiddleware,
     },
 };

libV2/tasks/BaseTask.js

@@ -1,11 +1,12 @@
+const assert = require('assert');
 const cron = require('node-schedule');
 const cronparser = require('cron-parser');
+const promClient = require('prom-client');
+const { DEFAULT_METRICS_ROUTE } = require('arsenal').network.probe.ProbeServer;
 
-const config = require('../config');
 const { client: cacheClient } = require('../cache');
 const Process = require('../process');
-const { LoggerContext } = require('../utils');
-const { Warp10Client } = require('../warp10');
+const { LoggerContext, iterIfError, startProbeServer } = require('../utils');
 
 const logger = new LoggerContext({
     module: 'BaseTask',
@@ -16,30 +17,104 @@ class Now {}
 class BaseTask extends Process {
     constructor(options) {
         super();
+        assert.notStrictEqual(options, undefined);
+        assert(Array.isArray(options.warp10), 'you must provide an array of warp 10 clients');
         this._cache = cacheClient;
-        this._warp10 = new Warp10Client({
-            ...config.warp10,
-            ...((options && options.warp10) || {}),
-        });
+        this._warp10Clients = options.warp10;
         this._scheduler = null;
         this._defaultSchedule = Now;
         this._defaultLag = 0;
-        this._nodeId = config.nodeId;
+        this._enableMetrics = options.enableMetrics || false;
+        this._metricsHost = options.metricsHost || 'localhost';
+        this._metricsPort = options.metricsPort || 9001;
+        this._metricsHandlers = null;
+        this._probeServer = null;
     }
 
-    async _setup() {
-        this._program
-            .option('-n, --now', 'Execute the task immediately and then exit. Overrides --schedule.')
-            .option(
-                '-s, --schedule <crontab>',
-                'Execute task using this crontab.  Overrides configured schedule',
-                value => {
-                    cronparser.parseExpression(value);
-                    return value;
-                },
-            )
-            .option('-l, --lag <lag>', 'Set a custom lag time in seconds', v => parseInt(v, 10))
-            .option('-n, --node-id <id>', 'Set a custom node id');
+    async _setup(includeDefaultOpts = true) {
+        if (includeDefaultOpts) {
+            this._program
+                .option('-n, --now', 'Execute the task immediately and then exit. Overrides --schedule.')
+                .option(
+                    '-s, --schedule <crontab>',
+                    'Execute task using this crontab.  Overrides configured schedule',
+                    value => {
+                        cronparser.parseExpression(value);
+                        return value;
+                    },
+                )
+                .option('-l, --lag <lag>', 'Set a custom lag time in seconds', v => parseInt(v, 10))
+                .option('-n, --node-id <id>', 'Set a custom node id');
+        }
+
+        if (this._enableMetrics) {
+            promClient.collectDefaultMetrics({
+                timeout: 10000,
+                gcDurationBuckets: [0.001, 0.01, 0.1, 1, 2, 5],
+            });
+
+            this._metricsHandlers = {
+                ...this._registerDefaultMetricHandlers(),
+                ...this._registerMetricHandlers(),
+            };
+            await this._createProbeServer();
+        }
+    }
+
+    _registerDefaultMetricHandlers() {
+        const taskName = this.constructor.name;
+
+        // Get the name of our subclass in snake case format eg BaseClass => _base_class
+        const taskNameSnake = taskName.replace(/[A-Z]/g, letter => `_${letter.toLowerCase()}`);
+
+        const executionDuration = new promClient.Gauge({
+            name: `s3_utapi${taskNameSnake}_duration_seconds`,
+            help: `Execution time of the ${taskName} task`,
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const executionAttempts = new promClient.Counter({
+            name: `s3_utapi${taskNameSnake}_attempts_total`,
+            help: `Total number of attempts to execute the ${taskName} task`,
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const executionFailures = new promClient.Counter({
+            name: `s3_utapi${taskNameSnake}_failures_total`,
+            help: `Total number of failures executing the ${taskName} task`,
+            labelNames: ['origin', 'containerName'],
+        });
+
+        return {
+            executionDuration,
+            executionAttempts,
+            executionFailures,
+        };
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        return {};
+    }
+
+    async _createProbeServer() {
+        this._probeServer = await startProbeServer({
+            bindAddress: this._metricsHost,
+            port: this._metricsPort,
+        });
+
+        this._probeServer.addHandler(
+            DEFAULT_METRICS_ROUTE,
+            (res, log) => {
+                log.debug('metrics requested');
+                res.writeHead(200, {
+                    'Content-Type': promClient.register.contentType,
+                });
+                promClient.register.metrics().then(metrics => {
+                    res.end(metrics);
+                });
+            },
+        );
     }
 
     get schedule() {
@@ -59,13 +134,6 @@ class BaseTask extends Process {
         return this._defaultLag;
     }
 
-    get nodeId() {
-        if (this._program.nodeId) {
-            return this._program.nodeId;
-        }
-        return this._nodeId;
-    }
-
     async _start() {
         await this._cache.connect();
         if (this.schedule === Now) {
@@ -87,12 +155,23 @@ class BaseTask extends Process {
     }
 
     async execute() {
+        let endTimer;
+        if (this._enableMetrics) {
+            endTimer = this._metricsHandlers.executionDuration.startTimer();
+            this._metricsHandlers.executionAttempts.inc(1);
+        }
+
         try {
             const timestamp = new Date() * 1000; // Timestamp in microseconds;
             const laggedTimestamp = timestamp - (this.lag * 1000000);
             await this._execute(laggedTimestamp);
         } catch (error) {
             logger.error('Error during task execution', { error });
+            this._metricsHandlers.executionFailures.inc(1);
+        }
+
+        if (this._enableMetrics) {
+            endTimer();
         }
     }
 
@@ -102,8 +181,28 @@ class BaseTask extends Process {
     }
 
     async _join() {
+        if (this._probeServer !== null) {
+            this._probeServer.stop();
+        }
         return this._cache.disconnect();
     }
+
+    withWarp10(func, onError) {
+        return iterIfError(this._warp10Clients, func, error => {
+            if (onError) {
+                onError(error);
+            } else {
+                const {
+                    name, code, message, stack,
+                } = error;
+                logger.error('error during warp 10 request', {
+                    error: {
+                        name, code, errmsg: message, stack: name !== 'RequestError' ? stack : undefined,
+                    },
+                });
+            }
+        });
+    }
 }
 
 module.exports = BaseTask;

libV2/tasks/CreateCheckpoint.js

@@ -1,3 +1,4 @@
+const promClient = require('prom-client');
 const BaseTask = require('./BaseTask');
 const config = require('../config');
 const { checkpointLagSecs, indexedEventFields } = require('../constants');
@@ -10,30 +11,103 @@ const logger = new LoggerContext({
 class CreateCheckpoint extends BaseTask {
     constructor(options) {
         super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.checkpointPort,
             ...options,
         });
+
         this._defaultSchedule = config.checkpointSchedule;
         this._defaultLag = checkpointLagSecs;
     }
 
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        const created = new promClient.Counter({
+            name: 's3_utapi_create_checkpoint_created_total',
+            help: 'Total number of checkpoints created',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const getLastCheckpoint = this._getLastCheckpoint.bind(this);
+        const lastCheckpoint = new promClient.Gauge({
+            name: 's3_utapi_create_checkpoint_last_checkpoint_seconds',
+            help: 'Timestamp of the last successfully created checkpoint',
+            labelNames: ['origin', 'containerName'],
+            async collect() {
+                try {
+                    const timestamp = await getLastCheckpoint();
+                    if (timestamp !== null) {
+                        this.set(timestamp);
+                    }
+                } catch (error) {
+                    logger.error('error during metric collection', { error });
+                }
+            },
+        });
+
+        return {
+            created,
+            lastCheckpoint,
+        };
+    }
+
+    /**
+     * Metrics for CreateCheckpoint
+     * @typedef {Object} CreateCheckpointMetrics
+     * @property {number} created - Number of checkpoints created
+     */
+
+    /**
+     *
+     * @param {CreateCheckpointMetrics} metrics - Metric values to push
+     * @returns {undefined}
+     */
+    _pushMetrics(metrics) {
+        if (!this._enableMetrics) {
+            return;
+        }
+
+        if (metrics.created !== undefined) {
+            this._metricsHandlers.created.inc(metrics.created);
+        }
+    }
+
+    async _getLastCheckpoint() {
+        const resp = await this.withWarp10(async warp10 => warp10.fetch({
+            className: 'utapi.checkpoint.master',
+            labels: {
+                node: warp10.nodeId,
+            },
+            start: 'now',
+            stop: -1,
+        }));
+
+        if (!resp.result || (resp.result.length === 0 || resp.result[0] === '' || resp.result[0] === '[]')) {
+            return null;
+        }
+
+        const result = JSON.parse(resp.result[0])[0];
+        const timestamp = result.v[0][0];
+        return timestamp / 1000000;// Convert timestamp from microseconds to seconds
+    }
+
     async _execute(timestamp) {
         logger.debug('creating checkpoints', { checkpointTimestamp: timestamp });
-
-        const params = {
-            params: {
-                nodeId: this.nodeId,
-                end: timestamp.toString(),
-                fields: indexedEventFields,
-            },
-            macro: 'utapi/createCheckpoint',
-        };
-        const status = await this._warp10.exec(params);
+        const status = await this.withWarp10(async warp10 => {
+            const params = {
+                params: {
+                    nodeId: warp10.nodeId,
+                    end: timestamp.toString(),
+                    fields: indexedEventFields,
+                },
+                macro: 'utapi/createCheckpoint',
+            };
+            return warp10.exec(params);
+        });
         if (status.result[0]) {
             logger.info(`created ${status.result[0] || 0} checkpoints`);
+            this._pushMetrics({ created: status.result[0] });
         }
     }
 }

libV2/tasks/CreateSnapshot.js

@@ -1,3 +1,4 @@
+const promClient = require('prom-client');
 const BaseTask = require('./BaseTask');
 const config = require('../config');
 const { snapshotLagSecs } = require('../constants');
@@ -10,29 +11,103 @@ const logger = new LoggerContext({
 class CreateSnapshot extends BaseTask {
     constructor(options) {
         super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.snapshotPort,
             ...options,
         });
+
         this._defaultSchedule = config.snapshotSchedule;
         this._defaultLag = snapshotLagSecs;
     }
 
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        const created = new promClient.Counter({
+            name: 's3_utapi_create_snapshot_created_total',
+            help: 'Total number of snapshots created',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const getLastSnapshot = this._getLastSnapshot.bind(this);
+        const lastSnapshot = new promClient.Gauge({
+            name: 's3_utapi_create_snapshot_last_snapshot_seconds',
+            help: 'Timestamp of the last successfully created snapshot',
+            labelNames: ['origin', 'containerName'],
+            async collect() {
+                try {
+                    const timestamp = await getLastSnapshot();
+                    if (timestamp !== null) {
+                        this.set(timestamp);
+                    }
+                } catch (error) {
+                    logger.error('error during metric collection', { error });
+                }
+            },
+        });
+
+        return {
+            created,
+            lastSnapshot,
+        };
+    }
+
+    /**
+     * Metrics for CreateSnapshot
+     * @typedef {Object} CreateSnapshotMetrics
+     * @property {number} created - Number of snapshots created
+     */
+
+    /**
+     *
+     * @param {CreateSnapshotMetrics} metrics - Metric values to push
+     * @returns {undefined}
+     */
+    _pushMetrics(metrics) {
+        if (!this._enableMetrics) {
+            return;
+        }
+
+        if (metrics.created !== undefined) {
+            this._metricsHandlers.created.inc(metrics.created);
+        }
+    }
+
+    async _getLastSnapshot() {
+        const resp = await this.withWarp10(async warp10 => warp10.fetch({
+            className: 'utapi.snapshot.master',
+            labels: {
+                node: warp10.nodeId,
+            },
+            start: 'now',
+            stop: -1,
+        }));
+
+        if (!resp.result || (resp.result.length === 0 || resp.result[0] === '' || resp.result[0] === '[]')) {
+            return null;
+        }
+
+        const result = JSON.parse(resp.result[0])[0];
+        const timestamp = result.v[0][0];
+        return timestamp / 1000000;// Convert timestamp from microseconds to seconds
+    }
+
     async _execute(timestamp) {
         logger.debug('creating snapshots', { snapshotTimestamp: timestamp });
 
-        const params = {
-            params: {
-                nodeId: this.nodeId,
-                end: timestamp.toString(),
-            },
-            macro: 'utapi/createSnapshot',
-        };
-        const status = await this._warp10.exec(params);
+        const status = await this.withWarp10(async warp10 => {
+            const params = {
+                params: {
+                    nodeId: warp10.nodeId,
+                    end: timestamp.toString(),
+                },
+                macro: 'utapi/createSnapshot',
+            };
+            return warp10.exec(params);
+        });
         if (status.result[0]) {
             logger.info(`created ${status.result[0]} snapshots`);
+            this._pushMetrics({ created: status.result[0] });
         }
     }
 }

libV2/tasks/DiskUsage.js

@@ -0,0 +1,300 @@
+const async = require('async');
+const Path = require('path');
+const fs = require('fs');
+const promClient = require('prom-client');
+const BaseTask = require('./BaseTask');
+const config = require('../config');
+const { expirationChunkDuration } = require('../constants');
+const {
+    LoggerContext, getFolderSize, formatDiskSize, sliceTimeRange,
+} = require('../utils');
+
+const moduleLogger = new LoggerContext({
+    module: 'MonitorDiskUsage',
+    path: config.diskUsage.path,
+});
+
+const WARN_THRESHOLD = 0.8;
+const ACTION_THRESHOLD = 0.95;
+
+class MonitorDiskUsage extends BaseTask {
+    constructor(options) {
+        super({
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.diskUsagePort,
+            ...options,
+        });
+
+        this._defaultSchedule = config.diskUsageSchedule;
+        this._defaultLag = 0;
+        this._path = config.diskUsage.path;
+        this._enabled = config.diskUsage.enabled;
+        this._expirationEnabled = config.diskUsage.expirationEnabled;
+        this._metricRetentionMicroSecs = config.diskUsage.retentionDays * 24 * 60 * 60 * 1000000;
+        this._hardLimit = config.diskUsage.hardLimit || null;
+    }
+
+    async _setup() {
+        await super._setup();
+        this._program
+            .option('--leader', 'Mark this process as the leader for metric expiration.')
+            .option(
+                '--lock',
+                'Manually trigger a lock of the warp 10 database. This will cause all other options to be ignored.',
+            )
+            .option(
+                '--unlock',
+                'Manually trigger an unlock of the warp 10 database. This will cause all other options to be ignored.',
+            );
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        const isLocked = new promClient.Gauge({
+            name: 's3_utapi_monitor_disk_usage_is_locked',
+            help: 'Indicates whether the monitored warp 10 has had writes disabled',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const leveldbBytes = new promClient.Gauge({
+            name: 's3_utapi_monitor_disk_usage_leveldb_bytes',
+            help: 'Total bytes used by warp 10 leveldb',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const datalogBytes = new promClient.Gauge({
+            name: 's3_utapi_monitor_disk_usage_datalog_bytes',
+            help: 'Total bytes used by warp 10 datalog',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const hardLimitRatio = new promClient.Gauge({
+            name: 's3_utapi_monitor_disk_usage_hard_limit_ratio',
+            help: 'Percent of the hard limit used by warp 10',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const hardLimitSetting = new promClient.Gauge({
+            name: 's3_utapi_monitor_disk_usage_hard_limit_bytes',
+            help: 'The hard limit setting in bytes',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        return {
+            isLocked,
+            leveldbBytes,
+            datalogBytes,
+            hardLimitRatio,
+            hardLimitSetting,
+        };
+    }
+
+    /**
+     * Metrics for MonitorDiskUsage
+     * @typedef {Object} MonitorDiskUsageMetrics
+     * @property {boolean} isLocked - Indicates if writes have been disabled for the monitored warp10
+     * @property {number} leveldbBytes - Total bytes used by warp 10 leveldb
+     * @property {number} datalogBytes - Total bytes used by warp 10 datalog
+     * @property {number} hardLimitRatio - Percent of the hard limit used by warp 10
+     * @property {number} hardLimitSetting - The hard limit setting in bytes
+     */
+
+    /**
+     *
+     * @param {MonitorDiskUsageMetrics} metrics - Metric values to push
+     * @returns {undefined}
+     */
+    _pushMetrics(metrics) {
+        if (!this._enableMetrics) {
+            return;
+        }
+
+        if (metrics.isLocked !== undefined) {
+            this._metricsHandlers.isLocked.set(metrics.isLocked ? 1 : 0);
+        }
+
+        if (metrics.leveldbBytes !== undefined) {
+            this._metricsHandlers.leveldbBytes.set(metrics.leveldbBytes);
+        }
+
+        if (metrics.datalogBytes !== undefined) {
+            this._metricsHandlers.datalogBytes.set(metrics.datalogBytes);
+        }
+
+        if (metrics.hardLimitRatio !== undefined) {
+            this._metricsHandlers.hardLimitRatio.set(metrics.hardLimitRatio);
+        }
+
+        if (metrics.hardLimitSetting !== undefined) {
+            this._metricsHandlers.hardLimitSetting.set(metrics.hardLimitSetting);
+        }
+    }
+
+    get isLeader() {
+        return this._program.leader !== undefined;
+    }
+
+    get isManualUnlock() {
+        return this._program.unlock !== undefined;
+    }
+
+    get isManualLock() {
+        return this._program.lock !== undefined;
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    async _getUsage(path) {
+        moduleLogger.debug(`calculating disk usage for ${path}`);
+        if (!fs.existsSync(path)) {
+            throw Error(`failed to calculate usage for non-existent path ${path}`);
+        }
+        return getFolderSize(path);
+    }
+
+    async _expireMetrics(timestamp) {
+        const resp = await this.withWarp10(async warp10 =>
+            warp10.exec({
+                macro: 'utapi/findOldestRecord',
+                params: {
+                    class: '~.*',
+                    labels: {},
+                },
+            }));
+
+        if (!resp.result || resp.result.length !== 1) {
+            moduleLogger.error('failed to fetch oldest record timestamp. expiration failed');
+            return;
+        }
+
+        const oldestTimestamp = resp.result[0];
+        if (oldestTimestamp === -1) {
+            moduleLogger.info('No records found, nothing to delete.');
+            return;
+        }
+
+        const endTimestamp = timestamp - this._metricRetentionMicroSecs;
+        if (oldestTimestamp > endTimestamp) {
+            moduleLogger.info('No records exceed retention period, nothing to delete.');
+            return;
+        }
+
+        await async.eachSeries(
+            sliceTimeRange(oldestTimestamp - 1, endTimestamp, expirationChunkDuration),
+            async ([start, end]) => {
+                moduleLogger.info('deleting metrics',
+                    { start, end });
+                return this.withWarp10(async warp10 =>
+                    warp10.delete({
+                        className: '~.*',
+                        start,
+                        end,
+                    }));
+            },
+        );
+    }
+
+    _checkHardLimit(size, nodeId) {
+        const hardPercentage = parseFloat((size / this._hardLimit).toFixed(2));
+        const hardLimitHuman = formatDiskSize(this._hardLimit);
+        const hardLogger = moduleLogger.with({
+            size,
+            sizeHuman: formatDiskSize(size),
+            hardPercentage,
+            hardLimit: this._hardLimit,
+            hardLimitHuman,
+            nodeId,
+        });
+
+        this._pushMetrics({ hardLimitRatio: hardPercentage });
+
+        const msg = `Using ${hardPercentage * 100}% of the ${hardLimitHuman} hard limit on ${nodeId}`;
+
+        if (hardPercentage < WARN_THRESHOLD) {
+            hardLogger.debug(msg);
+        } else if (hardPercentage >= WARN_THRESHOLD && hardPercentage < ACTION_THRESHOLD) {
+            hardLogger.warn(msg);
+        } else {
+            hardLogger.error(msg);
+            return true;
+        }
+        return false;
+    }
+
+    async _disableWarp10Updates() {
+        return this.withWarp10(async warp10 =>
+            warp10.exec({
+                script: `
+                DROP DROP
+                'Hard limit has been reached. Further updates have been disabled.'
+                'scality'
+                UPDATEOFF`,
+                params: {},
+            }));
+    }
+
+    async _enableWarp10Updates() {
+        return this.withWarp10(async warp10 =>
+            warp10.exec({
+                script: "DROP DROP 'scality' UPDATEON",
+                params: {},
+            }));
+    }
+
+    async _execute(timestamp) {
+        if (this.isManualUnlock) {
+            moduleLogger.info('manually unlocking warp 10', { nodeId: this.nodeId });
+            await this._enableWarp10Updates();
+            this._pushMetrics({ isLocked: false });
+            return;
+        }
+
+        if (this.isManualLock) {
+            moduleLogger.info('manually locking warp 10', { nodeId: this.nodeId });
+            await this._disableWarp10Updates();
+            this._pushMetrics({ isLocked: true });
+            return;
+        }
+
+        if (this._expirationEnabled && this.isLeader) {
+            moduleLogger.info(`expiring metrics older than ${config.diskUsage.retentionDays} days`);
+            await this._expireMetrics(timestamp);
+            return;
+        }
+
+        if (!this._enabled) {
+            moduleLogger.debug('disk usage monitoring not enabled, skipping check');
+            return;
+        }
+
+        let leveldbBytes = null;
+        let datalogBytes = null;
+        try {
+            leveldbBytes = await this._getUsage(Path.join(this._path, 'leveldb'));
+            datalogBytes = await this._getUsage(Path.join(this._path, 'datalog'));
+        } catch (error) {
+            moduleLogger.error(`error calculating disk usage for ${this._path}`, { error });
+            return;
+        }
+
+        this._pushMetrics({ leveldbBytes, datalogBytes });
+
+        const size = leveldbBytes + datalogBytes;
+        if (this._hardLimit !== null) {
+            moduleLogger.info(`warp 10 using ${formatDiskSize(size)} of disk space`, { leveldbBytes, datalogBytes });
+
+            const shouldLock = this._checkHardLimit(size, this.nodeId);
+            if (shouldLock) {
+                moduleLogger.warn('hard limit exceeded, disabling writes to warp 10', { nodeId: this.nodeId });
+                await this._disableWarp10Updates();
+            } else {
+                moduleLogger.info('usage below hard limit, ensuring writes to warp 10 are enabled',
+                    { nodeId: this.nodeId });
+                await this._enableWarp10Updates();
+            }
+            this._pushMetrics({ isLocked: shouldLock, hardLimitSetting: this._hardLimit });
+        }
+    }
+}
+
+module.exports = MonitorDiskUsage;

libV2/tasks/Downsample.js

@@ -1,41 +0,0 @@
-const BaseTask = require('./BaseTask');
-const config = require('../config');
-const { LoggerContext } = require('../utils');
-const { downsampleLagSecs, indexedEventFields } = require('../constants');
-
-const logger = new LoggerContext({
-    module: 'Repair',
-});
-
-class DownsampleTask extends BaseTask {
-    constructor(options) {
-        super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
-            ...options,
-        });
-        this._defaultSchedule = config.downsampleSchedule;
-        this._defaultLag = downsampleLagSecs;
-    }
-
-    async _execute(timestamp) {
-        logger.debug('Downsampling records', { timestamp, nodeId: this.nodeId });
-
-        const params = {
-            params: {
-                nodeId: this.nodeId,
-                end: timestamp.toString(),
-                fields: indexedEventFields,
-            },
-            macro: 'utapi/repairRecords',
-        };
-        const status = await this._warp10.exec(params);
-        if (status.result[0]) {
-            logger.info(`created ${status.result[0]} corrections`);
-        }
-    }
-}
-
-module.exports = DownsampleTask;

libV2/tasks/IngestShard.js

@@ -1,24 +1,113 @@
 const assert = require('assert');
+const async = require('async');
+const promClient = require('prom-client');
 const BaseTask = require('./BaseTask');
 const { UtapiMetric } = require('../models');
 const config = require('../config');
-const {
-    LoggerContext, shardFromTimestamp, convertTimestamp, InterpolatedClock,
-} = require('../utils');
 const { checkpointLagSecs } = require('../constants');
+const {
+    LoggerContext, shardFromTimestamp, convertTimestamp, InterpolatedClock, now,
+} = require('../utils');
 
 const logger = new LoggerContext({
     module: 'IngestShard',
 });
 
-const now = () => convertTimestamp(new Date().getTime());
 const checkpointLagMicroseconds = convertTimestamp(checkpointLagSecs);
 
 class IngestShardTask extends BaseTask {
-    constructor(...options) {
-        super(...options);
+    constructor(options) {
+        super({
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.ingestPort,
+            ...options,
+        });
+
         this._defaultSchedule = config.ingestionSchedule;
         this._defaultLag = config.ingestionLagSeconds;
+        this._stripEventUUID = options.stripEventUUID !== undefined ? options.stripEventUUID : true;
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        const ingestedTotal = new promClient.Counter({
+            name: 's3_utapi_ingest_shard_task_ingest_total',
+            help: 'Total number of metrics ingested',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const ingestedSlow = new promClient.Counter({
+            name: 's3_utapi_ingest_shard_task_slow_total',
+            help: 'Total number of slow metrics ingested',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const ingestedShards = new promClient.Counter({
+            name: 's3_utapi_ingest_shard_task_shard_ingest_total',
+            help: 'Total number of metric shards ingested',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const shardAgeTotal = new promClient.Counter({
+            name: 's3_utapi_ingest_shard_task_shard_age_total',
+            help: 'Total aggregated age of shards',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        return {
+            ingestedTotal,
+            ingestedSlow,
+            ingestedShards,
+            shardAgeTotal,
+        };
+    }
+
+    /**
+     * Metrics for IngestShardTask
+     * @typedef {Object} IngestShardMetrics
+     * @property {number} ingestedTotal - Number of events ingested
+     * @property {number} ingestedSlow - Number of slow events ingested
+     * @property {number} ingestedShards - Number of metric shards ingested
+     * @property {number} shardAgeTotal - Aggregated age of shards
+     */
+
+    /**
+     *
+     * @param {IngestShardMetrics} metrics - Metric values to push
+     * @returns {undefined}
+     */
+    _pushMetrics(metrics) {
+        if (!this._enableMetrics) {
+            return;
+        }
+
+        if (metrics.ingestedTotal !== undefined) {
+            this._metricsHandlers.ingestedTotal.inc(metrics.ingestedTotal);
+        }
+
+        if (metrics.ingestedSlow !== undefined) {
+            this._metricsHandlers.ingestedSlow.inc(metrics.ingestedSlow);
+        }
+
+        if (metrics.ingestedShards !== undefined) {
+            this._metricsHandlers.ingestedShards.inc(metrics.ingestedShards);
+        }
+
+        if (metrics.shardAgeTotal !== undefined) {
+            this._metricsHandlers.shardAgeTotal.inc(metrics.shardAgeTotal);
+        }
+    }
+
+    _hydrateEvent(data, stripTimestamp = false) {
+        const event = JSON.parse(data);
+        if (this._stripEventUUID) {
+            delete event.uuid;
+        }
+        if (stripTimestamp) {
+            delete event.timestamp;
+        }
+        return new UtapiMetric(event);
     }
 
     async _execute(timestamp) {
@@ -35,41 +124,61 @@ class IngestShardTask extends BaseTask {
             return;
         }
 
-        await Promise.all(toIngest.map(
+        let shardAgeTotal = 0;
+        let ingestedShards = 0;
+        await async.eachLimit(toIngest, 10,
             async shard => {
                 if (await this._cache.shardExists(shard)) {
                     const metrics = await this._cache.getMetricsForShard(shard);
                     if (metrics.length > 0) {
                         logger.info(`Ingesting ${metrics.length} events from shard`, { shard });
                         const shardAge = now() - shard;
-                        let metricClass;
-                        let records;
-                        if (shardAge < checkpointLagMicroseconds) {
-                            metricClass = 'utapi.event';
-                            records = metrics
-                                .map(m => new UtapiMetric(JSON.parse(m)));
-                        } else {
+                        const areSlowEvents = shardAge >= checkpointLagMicroseconds;
+                        const metricClass = areSlowEvents ? 'utapi.repair.event' : 'utapi.event';
+
+                        if (areSlowEvents) {
                             logger.info('Detected slow records, ingesting as repair');
-                            metricClass = 'utapi.repair.event';
-                            const clock = new InterpolatedClock();
-                            records = metrics
-                                .map(data => {
-                                    const metric = JSON.parse(data);
-                                    metric.timestamp = clock.getTs();
-                                    return new UtapiMetric(metric);
-                                });
                         }
-                        const status = await this._warp10.ingest({ className: metricClass }, records);
+
+                        const records = metrics.map(m => this._hydrateEvent(m, areSlowEvents));
+
+                        records.sort((a, b) => a.timestamp - b.timestamp);
+
+                        const clock = new InterpolatedClock();
+                        records.forEach(r => {
+                            r.timestamp = clock.getTs(r.timestamp);
+                        });
+
+                        let ingestedIntoNodeId;
+                        const status = await this.withWarp10(async warp10 => {
+                            // eslint-disable-next-line prefer-destructuring
+                            ingestedIntoNodeId = warp10.nodeId;
+                            return warp10.ingest(
+                                {
+                                    className: metricClass,
+                                    labels: { origin: config.nodeId },
+                                }, records,
+                            );
+                        });
                         assert.strictEqual(status, records.length);
                         await this._cache.deleteShard(shard);
+                        logger.info(`ingested ${status} records from ${config.nodeId} into ${ingestedIntoNodeId}`);
+
+                        shardAgeTotal += shardAge;
+                        ingestedShards += 1;
+                        this._pushMetrics({ ingestedTotal: records.length });
+                        if (areSlowEvents) {
+                            this._pushMetrics({ ingestedSlow: records.length });
+                        }
                     } else {
                         logger.debug('No events found in shard, cleaning up');
                     }
                 } else {
                     logger.warn('shard does not exist', { shard });
                 }
-            },
-        ));
+            });
+        const shardAgeTotalSecs = shardAgeTotal / 1000000;
+        this._pushMetrics({ shardAgeTotal: shardAgeTotalSecs, ingestedShards });
     }
 }
 

libV2/tasks/ManualAdjust.js

@@ -0,0 +1,85 @@
+const async = require('async');
+const BaseTask = require('./BaseTask');
+const UtapiClient = require('../client');
+const { LoggerContext } = require('../utils');
+
+const logger = new LoggerContext({
+    module: 'ManualAdjust',
+});
+
+function collectArgs(arg, prev) {
+    return prev.concat([arg]);
+}
+
+class ManualAdjust extends BaseTask {
+    async _setup() {
+        // Don't include default flags
+        await super._setup(false);
+        this._program
+            .option('-h, --host <host>', 'Utapi server host', 'localhost')
+            .option('-p, --port <port>', 'Utapi server port', '8100', parseInt)
+            .option('-b, --bucket <buckets...>', 'target these buckets', collectArgs, [])
+            .option('-a, --account <accounts...>', 'target these accounts', collectArgs, [])
+            .option('-u, --user <users...>', 'target these users', collectArgs, [])
+            .requiredOption('-o, --objects <adjustment>', 'adjust numberOfObjects by this amount', parseInt)
+            .requiredOption('-s, --storage <adjustment>', 'adjust storageUtilized by this amount', parseInt);
+    }
+
+    async _start() {
+        this._utapiClient = new UtapiClient({
+            host: this._program.host,
+            port: this._program.port,
+            disableRetryCache: true,
+        });
+        await super._start();
+    }
+
+    async _pushAdjustmentMetric(metric) {
+        logger.info('pushing adjustment metric', { metric });
+        await this._utapiClient.pushMetric(metric);
+    }
+
+    async _execute() {
+        const timestamp = Date.now();
+
+        const objectDelta = this._program.objects;
+        const sizeDelta = this._program.storage;
+
+        if (!this._program.bucket.length && !this._program.account.length && !this._program.user.length) {
+            throw Error('You must provided at least one of --bucket, --account or --user');
+        }
+
+        logger.info('writing adjustments');
+        if (this._program.bucket.length) {
+            logger.info('adjusting buckets');
+            await async.eachSeries(
+                this._program.bucket,
+                async bucket => this._pushAdjustmentMetric({
+                    bucket, objectDelta, sizeDelta, timestamp,
+                }),
+            );
+        }
+
+        if (this._program.account.length) {
+            logger.info('adjusting accounts');
+            await async.eachSeries(
+                this._program.account,
+                async account => this._pushAdjustmentMetric({
+                    account, objectDelta, sizeDelta, timestamp,
+                }),
+            );
+        }
+
+        if (this._program.user.length) {
+            logger.info('adjusting users');
+            await async.eachSeries(
+                this._program.user,
+                async user => this._pushAdjustmentMetric({
+                    user, objectDelta, sizeDelta, timestamp,
+                }),
+            );
+        }
+    }
+}
+
+module.exports = ManualAdjust;

libV2/tasks/Migrate.js

@@ -6,7 +6,12 @@ const { UtapiRecord } = require('../models');
 const config = require('../config');
 const errors = require('../errors');
 const RedisClient = require('../redis');
-const { warp10RecordType, operations: operationIds, serviceToWarp10Label } = require('../constants');
+const {
+    warp10RecordType,
+    operations: operationIds,
+    serviceToWarp10Label,
+    migrationOpTranslationMap,
+} = require('../constants');
 const {
     LoggerContext,
     now,
@@ -37,14 +42,7 @@ const LEVELS_TO_MIGRATE = [
 
 class MigrateTask extends BaseTask {
     constructor(options) {
-        super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
-            ...options,
-        });
-
+        super(options);
         this._failedCorrections = [];
         this._redis = new RedisClient(config.redis);
     }
@@ -141,7 +139,20 @@ class MigrateTask extends BaseTask {
                 timestamp,
                 timestamp,
             ));
-            const numberOfObjects = MigrateTask._parseMetricValue(numberOfObjectsResp[0]);
+
+            let numberOfObjects;
+            if (numberOfObjectsResp.length === 1) {
+                numberOfObjects = MigrateTask._parseMetricValue(numberOfObjectsResp[0]);
+            } else {
+                numberOfObjects = numberOfObjectsOffset;
+                logger.warn('Could not retrieve value for numberOfObjects, falling back to last seen value',
+                    {
+                        metricLevel: level,
+                        resource,
+                        metricTimestamp: timestamp,
+                        lastSeen: numberOfObjectsOffset,
+                    });
+            }
 
             let incomingBytes = 0;
             let outgoingBytes = 0;
@@ -154,6 +165,8 @@ class MigrateTask extends BaseTask {
                     outgoingBytes = apiOp.count;
                 } else if (operationIds.includes(apiOp.op)) {
                     operations[apiOp.op] = apiOp.count;
+                } else if (migrationOpTranslationMap[apiOp.op] !== undefined) {
+                    operations[migrationOpTranslationMap[apiOp.op]] = apiOp.count;
                 } else {
                     logger.warn('dropping unknown operation', { apiOp });
                 }
@@ -174,16 +187,16 @@ class MigrateTask extends BaseTask {
     }
 
     async _findLatestSnapshot(level, resource) {
-        const resp = await this._warp10.fetch({
+        const resp = await this.withWarp10(async warp10 => warp10.fetch({
             className: 'utapi.snapshot',
             labels: {
                 [serviceToWarp10Label[level]]: resource,
             },
             start: 'now',
             stop: -1,
-        });
+        }));
 
-        if (resp.result && (resp.result.length === 0 || resp.result[0] === '')) {
+        if (resp.result && (resp.result.length === 0 || resp.result[0] === '' || resp.result[0] === '[]')) {
             return null;
         }
 
@@ -195,14 +208,14 @@ class MigrateTask extends BaseTask {
         let pos = beginTimestamp;
         // eslint-disable-next-line no-constant-condition
         while (true) {
-            const resp = await this._warp10.fetch({
+            const resp = await this.withWarp10(async warp10 => warp10.fetch({
                 className: 'utapi.snapshot',
                 labels: {
                     [serviceToWarp10Label[level]]: resource,
                 },
                 start: pos - 1,
                 stop: -WARP10_SCAN_SIZE,
-            });
+            }));
             if (resp.result && resp.result.length === 0) {
                 return pos;
             }
@@ -218,7 +231,7 @@ class MigrateTask extends BaseTask {
     }
 
     async _migrateMetric(className, level, resource, metric) {
-        return this._warp10.ingest(
+        return this.withWarp10(async warp10 => warp10.ingest(
             {
                 className,
                 labels: {
@@ -227,7 +240,7 @@ class MigrateTask extends BaseTask {
                 valueType: warp10RecordType,
             },
             [metric],
-        );
+        ));
     }
 
     static _sumRecord(a, b) {

libV2/tasks/Reindex.js

@@ -7,7 +7,12 @@ const config = require('../config');
 const metadata = require('../metadata');
 const { serviceToWarp10Label, warp10RecordType } = require('../constants');
 
-const { LoggerContext, convertTimestamp } = require('../utils');
+const {
+    LoggerContext,
+    logEventFilter,
+    convertTimestamp,
+    buildFilterChain,
+} = require('../utils');
 
 const logger = new LoggerContext({
     module: 'ReindexTask',
@@ -16,14 +21,35 @@ const logger = new LoggerContext({
 class ReindexTask extends BaseTask {
     constructor(options) {
         super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.reindexPort,
             ...options,
         });
+
         this._defaultSchedule = config.reindexSchedule;
         this._defaultLag = 0;
+        const eventFilters = (config && config.filter) || {};
+        this._shouldReindex = buildFilterChain((config && config.filter) || {});
+        // exponential backoff: max wait = 50 * 2 ^ 10 milliseconds ~= 51 seconds
+        this.ebConfig = {
+            times: 10,
+            interval: retryCount => 50 * (2 ** retryCount),
+        };
+
+        if (Object.keys(eventFilters).length !== 0) {
+            logEventFilter((...args) => logger.info(...args), 'reindex resource filtering enabled', eventFilters);
+        }
+    }
+
+    async _setup(includeDefaultOpts = true) {
+        await super._setup(includeDefaultOpts);
+        this._program.option(
+            '--bucket <bucket>',
+            'Manually specify a bucket to reindex. Can be used multiple times.',
+            (bucket, previous) => previous.concat([bucket]),
+            [],
+        );
     }
 
     static async _indexBucket(bucket) {
@@ -33,16 +59,23 @@ class ReindexTask extends BaseTask {
         let lastMasterSize = null;
 
         for await (const obj of metadata.listObjects(bucket)) {
-            if (obj.value.isDeleteMarker) {
+            if (obj.value.isDeleteMarker || obj.value.isPHD) {
                 // eslint-disable-next-line no-continue
                 continue;
             }
+
+            if (!Number.isInteger(obj.value['content-length'])) {
+                logger.debug('object missing content-length, not including in count');
+                // eslint-disable-next-line no-continue
+                continue;
+            }
+
             count += 1;
             size += obj.value['content-length'];
 
             // If versioned, subtract the size of the master to avoid double counting
             if (lastMaster && obj.name === lastMaster) {
-                logger.debug('Detected versioned key, subtracting master size', { lastMasterSize, key: obj.name });
+                logger.debug('Detected versioned key. subtracting master size', { lastMasterSize, key: obj.name });
                 size -= lastMasterSize;
                 count -= 1;
                 lastMaster = null;
@@ -67,29 +100,46 @@ class ReindexTask extends BaseTask {
 
     async _fetchCurrentMetrics(level, resource) {
         const timestamp = convertTimestamp(new Date().getTime());
-        const options = {
-            params: {
-                start: timestamp,
-                end: timestamp,
-                node: this._nodeId,
-                labels: {
-                    [level]: resource,
+        const res = await this.withWarp10(warp10 => {
+            const options = {
+                params: {
+                    end: timestamp,
+                    node: warp10.nodeId,
+                    labels: {
+                        [level]: resource,
+                    },
+                    // eslint-disable-next-line camelcase
+                    no_reindex: true,
                 },
-            },
-            macro: 'utapi/getMetrics',
+                macro: 'utapi/getMetricsAt',
+            };
+            return warp10.exec(options);
+        });
+
+        const [value] = res.result || [];
+        if (!value) {
+            throw new Error('unable to fetch current metrics from warp10');
+        }
+
+        if (!Number.isInteger(value.objD) || !Number.isInteger(value.sizeD)) {
+            logger.error('invalid values returned from warp 10', { response: res });
+            throw new Error('invalid values returned from warp 10');
+        }
+
+        return {
+            timestamp,
+            value,
         };
-        const res = await this._warp10.exec(options);
-        return { timestamp, value: JSON.parse(res.result[0]) };
     }
 
     async _updateMetric(level, resource, total) {
         const { timestamp, value } = await this._fetchCurrentMetrics(level, resource);
 
-        const objectDelta = total.count - value.numberOfObjects[0];
-        const sizeDelta = total.size - value.storageUtilized[0];
+        const objectDelta = total.count - value.objD;
+        const sizeDelta = total.size - value.sizeD;
 
         if (objectDelta !== 0 || sizeDelta !== 0) {
-            logger.info('discrepancy detected in metrics, writing corrective record',
+            logger.info('discrepancy detected in metrics. writing corrective record',
                 { [level]: resource, objectDelta, sizeDelta });
 
             const record = new UtapiRecord({
@@ -97,8 +147,7 @@ class ReindexTask extends BaseTask {
                 sizeDelta,
                 timestamp,
             });
-
-            await this._warp10.ingest(
+            await this.withWarp10(warp10 => warp10.ingest(
                 {
                     className: 'utapi.repair.reindex',
                     labels: {
@@ -107,29 +156,46 @@ class ReindexTask extends BaseTask {
                     valueType: warp10RecordType,
                 },
                 [record],
-            );
+            ));
         }
     }
 
+    get targetBuckets() {
+        if (this._program.bucket.length) {
+            return this._program.bucket.map(name => ({ name }));
+        }
+        return metadata.listBuckets();
+    }
+
     async _execute() {
-        logger.debug('reindexing objects');
+        logger.info('started reindex task');
 
         const accountTotals = {};
         const ignoredAccounts = new Set();
+        await async.eachLimit(this.targetBuckets, 5, async bucket => {
+            if (!this._shouldReindex({ bucket: bucket.name, account: bucket.account })) {
+                logger.debug('skipping excluded bucket', { bucket: bucket.name, account: bucket.account });
+                return;
+            }
 
-        await async.eachLimit(metadata.listBuckets(), 5, async bucket => {
-            logger.trace('starting reindex of bucket', { bucket: bucket.name });
+            logger.info('started bucket reindex', { bucket: bucket.name });
 
             const mpuBucket = `${mpuBucketPrefix}${bucket.name}`;
             let bktTotal;
             let mpuTotal;
 
             try {
-                bktTotal = await async.retryable(ReindexTask._indexBucket)(bucket.name);
-                mpuTotal = await async.retryable(ReindexTask._indexMpuBucket)(mpuBucket);
+                bktTotal = await async.retryable(this.ebConfig, ReindexTask._indexBucket)(bucket.name);
+                mpuTotal = await async.retryable(this.ebConfig, ReindexTask._indexMpuBucket)(mpuBucket);
             } catch (error) {
-                logger.error('failed to reindex bucket, ignoring associated account', { error, bucket: bucket.name });
-                ignoredAccounts.add(bucket.account);
+                logger.error(
+                    'failed bucket reindex. any associated account will be skipped',
+                    { error, bucket: bucket.name },
+                );
+                // buckets passed with `--bucket` won't have an account property
+                if (bucket.account) {
+                    ignoredAccounts.add(bucket.account);
+                }
                 return;
             }
 
@@ -138,33 +204,45 @@ class ReindexTask extends BaseTask {
                 count: bktTotal.count,
             };
 
-            if (accountTotals[bucket.account]) {
-                accountTotals[bucket.account].size += total.size;
-                accountTotals[bucket.account].count += total.count;
-            } else {
-                accountTotals[bucket.account] = { ...total };
+            // buckets passed with `--bucket` won't have an account property
+            if (bucket.account) {
+                if (accountTotals[bucket.account]) {
+                    accountTotals[bucket.account].size += total.size;
+                    accountTotals[bucket.account].count += total.count;
+                } else {
+                    accountTotals[bucket.account] = { ...total };
+                }
             }
 
-            logger.trace('finished indexing bucket', { bucket: bucket.name });
+            logger.info('finished bucket reindex', { bucket: bucket.name });
 
-            await this._updateMetric(
-                serviceToWarp10Label.buckets,
-                bucket.name,
-                total,
-            );
+            try {
+                await this._updateMetric(
+                    serviceToWarp10Label.buckets,
+                    bucket.name,
+                    total,
+                );
+            } catch (error) {
+                logger.error('error updating metrics for bucket', { error, bucket: bucket.name });
+            }
         });
 
         const toUpdate = Object.entries(accountTotals)
             .filter(([account]) => !ignoredAccounts.has(account));
 
-        await async.eachLimit(toUpdate, 5, async ([account, total]) =>
-            this._updateMetric(
-                serviceToWarp10Label.accounts,
-                account,
-                total,
-            ));
+        await async.eachLimit(toUpdate, 5, async ([account, total]) => {
+            try {
+                await this._updateMetric(
+                    serviceToWarp10Label.accounts,
+                    account,
+                    total,
+                );
+            } catch (error) {
+                logger.error('error updating metrics for account', { error, account });
+            }
+        });
 
-        logger.debug('finished reindexing');
+        logger.info('finished reindex task');
     }
 }
 

libV2/tasks/Repair.js

@@ -1,3 +1,4 @@
+const promClient = require('prom-client');
 const BaseTask = require('./BaseTask');
 const config = require('../config');
 const { LoggerContext } = require('../utils');
@@ -10,30 +11,67 @@ const logger = new LoggerContext({
 class RepairTask extends BaseTask {
     constructor(options) {
         super({
-            warp10: {
-                requestTimeout: 30000,
-                connectTimeout: 30000,
-            },
+            enableMetrics: config.metrics.enabled,
+            metricsHost: config.metrics.host,
+            metricsPort: config.metrics.repairPort,
             ...options,
         });
+
         this._defaultSchedule = config.repairSchedule;
         this._defaultLag = repairLagSecs;
     }
 
+    // eslint-disable-next-line class-methods-use-this
+    _registerMetricHandlers() {
+        const created = new promClient.Counter({
+            name: 's3_utapi_repair_task_created_total',
+            help: 'Total number of repair records created',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        return {
+            created,
+        };
+    }
+
+    /**
+     * Metrics for RepairTask
+     * @typedef {Object} RepairMetrics
+     * @property {number} created - Number of repair records created
+     */
+
+    /**
+     *
+     * @param {RepairMetrics} metrics - Metric values to push
+     * @returns {undefined}
+     */
+    _pushMetrics(metrics) {
+        if (!this._enableMetrics) {
+            return;
+        }
+
+        if (metrics.created !== undefined) {
+            this._metricsHandlers.created.inc(metrics.created);
+        }
+    }
+
     async _execute(timestamp) {
         logger.debug('Checking for repairs', { timestamp, nodeId: this.nodeId });
 
-        const params = {
-            params: {
-                nodeId: this.nodeId,
-                end: timestamp.toString(),
-                fields: indexedEventFields,
-            },
-            macro: 'utapi/repairRecords',
-        };
-        const status = await this._warp10.exec(params);
+        const status = await this.withWarp10(warp10 => {
+            const params = {
+                params: {
+                    nodeId: warp10.nodeId,
+                    end: timestamp.toString(),
+                    fields: indexedEventFields,
+                },
+                macro: 'utapi/repairRecords',
+            };
+            return warp10.exec(params);
+        });
         if (status.result[0]) {
             logger.info(`created ${status.result[0]} corrections`);
+            this._pushMetrics({ created: status.result[0] });
         }
     }
 }

libV2/tasks/index.js

@@ -5,6 +5,8 @@ const CreateSnapshot = require('./CreateSnapshot');
 const RepairTask = require('./Repair');
 const ReindexTask = require('./Reindex');
 const MigrateTask = require('./Migrate');
+const MonitorDiskUsage = require('./DiskUsage');
+const ManualAdjust = require('./ManualAdjust');
 
 module.exports = {
     IngestShard,
@@ -14,4 +16,6 @@ module.exports = {
     RepairTask,
     ReindexTask,
     MigrateTask,
+    MonitorDiskUsage,
+    ManualAdjust,
 };

libV2/utils/disk.js

@@ -0,0 +1,53 @@
+const { promisify } = require('util');
+const getFolderSize = require('get-folder-size');
+const byteSize = require('byte-size');
+
+const diskSpecRegex = /(\d+)([bkmgtpxz])(i?b)?/;
+const suffixToExp = {
+    b: 0,
+    k: 1,
+    m: 2,
+    g: 3,
+    t: 4,
+    p: 5,
+    x: 6,
+    z: 7,
+};
+
+/**
+ * Converts a string specifying disk size into its value in bytes
+ * Supported formats:
+ *  1b/1B - Directly specify a byte size
+ *  1K/1MB/1GiB - Specify a number of bytes using IEC or common suffixes
+ *
+ * Suffixes are case insensitive.
+ * All suffixes are considered IEC standard with 1 kibibyte being 2^10 bytes.
+ *
+ * @param {String} spec - string for conversion
+ * @returns {Integer} - disk size in bytes
+ */
+function parseDiskSizeSpec(spec) {
+    const normalized = spec.toLowerCase();
+    if (!diskSpecRegex.test(normalized)) {
+        throw Error('Format does not match a known suffix');
+    }
+
+    const match = diskSpecRegex.exec(normalized);
+    const size = parseInt(match[1], 10);
+    const exponent = suffixToExp[match[2]];
+    return size * (1024 ** exponent);
+}
+
+function _formatFunc() {
+    return `${this.value}${this.unit}`;
+}
+
+function formatDiskSize(value) {
+    return byteSize(value, { units: 'iec', toStringFn: _formatFunc }).toString();
+}
+
+module.exports = {
+    parseDiskSizeSpec,
+    getFolderSize: promisify(getFolderSize),
+    formatDiskSize,
+};

libV2/utils/filter.js

@@ -0,0 +1,47 @@
+const assert = require('assert');
+
+/**
+ * filterObject
+ *
+ * Constructs a function meant for filtering Objects by the value of a key
+ * Returned function returns a boolean with false meaning the object was present
+ * in the filter allowing the function to be passed directly to Array.filter etc.
+ *
+ * @param {string} key - Object key to inspect
+ * @param {Object} filter
+ * @param {Set} [filter.allow] - Set containing keys to include
+ * @param {Set} [filter.deny] - Set containing keys to not include
+ * @returns {function(Object): bool}
+ */
+
+function filterObject(obj, key, { allow, deny }) {
+    if (allow && deny) {
+        throw new Error('You can not define both an allow and a deny list.');
+    }
+    if (!allow && !deny) {
+        throw new Error('You must define either an allow or a deny list.');
+    }
+    if (allow) {
+        assert(allow instanceof Set);
+        return obj[key] === undefined || allow.has(obj[key]);
+    }
+    assert(deny instanceof Set);
+    return obj[key] === undefined || !deny.has(obj[key]);
+}
+
+/**
+ * buildFilterChain
+ *
+ * Constructs a function from a map of key names and allow/deny filters.
+ * The returned function returns a boolean with false meaning the object was present
+ * in one of the filters allowing the function to be passed directly to Array.filter etc.
+ *
+ * @param {Object<string, Object<string, Set>} filters
+ * @returns {function(Object): bool}
+ */
+
+function buildFilterChain(filters) {
+    return obj => Object.entries(filters).every(([key, filter]) => filterObject(obj, key, filter));
+}
+
+module.exports = { filterObject, buildFilterChain };

libV2/utils/func.js

@@ -30,7 +30,37 @@ function comprehend(data, func) {
     }, {});
 }
 
+/**
+ * Calls func with items in sequence, advancing if an error is thrown.
+ * The result from the first successful call is returned.
+ *
+ * onError, if passed, is called on every error thrown by func;
+ *
+ * @param {Array} items - items to iterate
+ * @param {AsyncFunction} func - function to apply to each item
+ * @param {Function|undefined} onError - optional function called if an error is thrown
+ * @returns {*} -
+ */
+async function iterIfError(items, func, onError) {
+    let error;
+    // eslint-disable-next-line no-restricted-syntax
+    for (const item of items) {
+        try {
+            // eslint-disable-next-line no-await-in-loop
+            const resp = await func(item);
+            return resp;
+        } catch (_error) {
+            if (onError) {
+                onError(_error);
+            }
+            error = _error;
+        }
+    }
+    throw error || new Error('unable to complete request');
+}
+
 module.exports = {
     asyncOrCallback,
     comprehend,
+    iterIfError,
 };

libV2/utils/index.js

@@ -2,10 +2,16 @@ const log = require('./log');
 const shard = require('./shard');
 const timestamp = require('./timestamp');
 const func = require('./func');
+const disk = require('./disk');
+const filter = require('./filter');
+const probe = require('./probe');
 
 module.exports = {
     ...log,
     ...shard,
     ...timestamp,
     ...func,
+    ...disk,
+    ...filter,
+    ...probe,
 };

libV2/utils/log.js

@@ -1,12 +1,5 @@
 const werelogs = require('werelogs');
-const config = require('../config');
-
-const loggerConfig = {
-    level: config.logging.level,
-    dump: config.logging.dumpLevel,
-};
-
-werelogs.configure(loggerConfig);
+const { comprehend } = require('./func');
 
 const rootLogger = new werelogs.Logger('Utapi');
 
@@ -77,8 +70,6 @@ class LoggerContext {
     }
 }
 
-rootLogger.debug('logger initialized', { loggerConfig });
-
 function buildRequestLogger(req) {
     let reqUids = [];
     if (req.headers['x-scal-request-uids'] !== undefined) {
@@ -102,8 +93,26 @@ function buildRequestLogger(req) {
     return new LoggerContext({}, reqLogger);
 }
 
+function logEventFilter(logger, msg, eventFilters) {
+    const filterLog = comprehend(
+        eventFilters,
+        (level, rules) => ({
+            key: level,
+            value: comprehend(
+                rules,
+                (rule, values) => ({
+                    key: rule,
+                    value: Array.from(values),
+                }),
+            ),
+        }),
+    );
+    logger(msg, { filters: filterLog });
+}
+
 module.exports = {
     logger: rootLogger,
     buildRequestLogger,
     LoggerContext,
+    logEventFilter,
 };

libV2/utils/probe.js

@@ -0,0 +1,32 @@
+
+const { ProbeServer } = require('arsenal').network.probe.ProbeServer;
+
+/**
+ * Configure probe servers
+ * @typedef {Object} ProbeServerConfig
+ * @property {string} bindAddress - Address to bind probe server to
+ * @property {number} port - Port to bind probe server to
+ */
+
+/**
+ * Start an empty probe server
+ * @async
+ * @param {ProbeServerConfig} config - Configuration for probe server
+ * @returns {Promise<ProbeServer>} - Instance of ProbeServer
+ */
+async function startProbeServer(config) {
+    if (!config) {
+        throw new Error('configuration for probe server is missing');
+    }
+
+    return new Promise((resolve, reject) => {
+        const probeServer = new ProbeServer(config);
+        probeServer.onListening(() => resolve(probeServer));
+        probeServer.onError(err => reject(err));
+        probeServer.start();
+    });
+}
+
+module.exports = {
+    startProbeServer,
+};

libV2/utils/timestamp.js

@@ -19,16 +19,16 @@ class InterpolatedClock {
         this._step = 1;
     }
 
-    getTs() {
-        const ts = new Date().now();
+    getTs(timestamp) {
+        const ts = timestamp !== undefined ? timestamp : Date.now();
         if (ts === this._now) {
             // If this is the same millisecond as the last call
             this._step += 1;
-            return ts * 1000 + (this._step - 1);
+            return convertTimestamp(ts) + (this._step - 1);
         }
         this._now = ts;
         this._step = 1;
-        return ts * 1000;
+        return convertTimestamp(ts);
     }
 }
 
@@ -42,8 +42,37 @@ function now() {
     return Date.now() * 1000;
 }
 
+/**
+ * Slice the time range represented by the passed timestamps
+ * into slices of at most `step` duration.
+ *
+ * Both `start` and `end` are included in the returned slices.
+ * Slice timestamps are inclusive and non overlapping.
+ *
+ * For example sliceTimeRange(0, 5, 2) will yield
+ * [0, 1]
+ * [2, 3]
+ * [4, 5]
+ *
+ * @param {Number} start
+ * @param {Number} end
+ * @param {Number} step
+ */
+
+function* sliceTimeRange(start, end, step) {
+    let spos = start;
+    let epos = start + step - 1;
+    while (epos < end) {
+        yield [spos, epos];
+        spos += step;
+        epos += step;
+    }
+    yield [spos, end];
+}
+
 module.exports = {
     convertTimestamp,
     InterpolatedClock,
     now,
+    sliceTimeRange,
 };

libV2/vault.js

@@ -1,151 +0,0 @@
-const assert = require('assert');
-const { auth, policies } = require('arsenal');
-const vaultclient = require('vaultclient');
-const config = require('./config');
-
-/**
-@class Vault
-* Creates a vault instance for authentication and authorization
-*/
-
-class Vault {
-    constructor(options) {
-        const { host, port } = options.vaultd;
-        if (options.https) {
-            const { key, cert, ca } = options.https;
-            this._client = new vaultclient.Client(host, port, true, key, cert,
-                ca);
-        } else {
-            this._client = new vaultclient.Client(host, port);
-        }
-    }
-
-    /** authenticateV4Request
-    *
-    * @param {object} params - the authentication parameters as returned by
-    *                          auth.extractParams
-    * @param {number} params.version - shall equal 4
-    * @param {string} params.data.accessKey - the user's accessKey
-    * @param {string} params.data.signatureFromRequest - the signature read from
-    *                                                    the request
-    * @param {string} params.data.region - the AWS region
-    * @param {string} params.data.stringToSign - the stringToSign
-    * @param {string} params.data.scopeDate - the timespan to allow the request
-    * @param {string} params.data.authType - the type of authentication
-    *   (query or header)
-    * @param {string} params.data.signatureVersion - the version of the
-    *                                                signature (AWS or AWS4)
-    * @param {number} params.data.signatureAge - the age of the signature in ms
-    * @param {string} params.data.log - the logger object
-     * @param {RequestContext []} requestContexts - an array of
-     * RequestContext instances which contain information
-     * for policy authorization check
-     * @param {function} callback - cb(err)
-     * @return {undefined}
-    */
-    authenticateV4Request(params, requestContexts, callback) {
-        const {
-            accessKey, signatureFromRequest, region, scopeDate,
-            stringToSign,
-        } = params.data;
-        const { log } = params;
-        log.debug('authenticating V4 request');
-        const serializedRCs = requestContexts.map(rc => rc.serialize());
-        this._client.verifySignatureV4(
-            stringToSign, signatureFromRequest,
-            accessKey, region, scopeDate,
-            { reqUid: log.getSerializedUids(), requestContext: serializedRCs },
-            (err, authInfo) => {
-                if (err) {
-                    log.trace('error from vault', { error: err });
-                    return callback(err);
-                }
-                return callback(null,
-                    authInfo.message.body.authorizationResults);
-            },
-        );
-    }
-
-    /**
-    * Returns canonical Ids for a given list of account Ids
-    * @param {string[]} accountIds - list of account ids
-    * @param {object} log - Werelogs request logger
-    * @return {Promise} -
-    */
-    getCanonicalIds(accountIds, log) {
-        log.debug('retrieving canonical ids for account ids', {
-            method: 'Vault.getCanonicalIds',
-        });
-        return new Promise((resolve, reject) =>
-            this._client.getCanonicalIdsByAccountIds(accountIds,
-                { reqUid: log.getSerializedUids(), logger: log }, (err, res) => {
-                    if (err) {
-                        reject(err);
-                        return;
-                    }
-                    resolve(res);
-                }));
-    }
-}
-
-const vault = new Vault(config);
-auth.setHandler(vault);
-
-async function authenticateRequest(request, action, level, resources) {
-    const policyContext = new policies.RequestContext(
-        request.headers,
-        request.query,
-        level,
-        resources,
-        request.ip,
-        request.ctx.encrypted,
-        action,
-        'utapi',
-    );
-
-    return new Promise((resolve, reject) => {
-        auth.server.doAuth(request, request.logger.logger, (err, res) => {
-            if (err && (err.InvalidAccessKeyId || err.AccessDenied)) {
-                resolve([false]);
-                return;
-            }
-            if (err) {
-                reject(err);
-                return;
-            }
-            // Will only have res if request is from a user rather than an account
-            if (res) {
-                try {
-                    const authorizedResources = (res || [])
-                        .reduce((authed, result) => {
-                            if (result.isAllowed) {
-                                // result.arn should be of format:
-                                // arn:scality:utapi:::resourcetype/resource
-                                assert(typeof result.arn === 'string');
-                                assert(result.arn.indexOf('/') > -1);
-                                const resource = result.arn.split('/')[1];
-                                authed.push(resource);
-                                request.logger.trace('access granted for resource', { resource });
-                            }
-                            return authed;
-                        }, []);
-                    resolve([
-                        authorizedResources.length !== 0,
-                        authorizedResources,
-                    ]);
-                } catch (err) {
-                    reject(err);
-                }
-            } else {
-                request.logger.trace('granted access to all resources');
-                resolve([true]);
-            }
-        }, 's3', [policyContext]);
-    });
-}
-
-module.exports = {
-    authenticateRequest,
-    Vault,
-    vault,
-};

libV2/vault/client.js

@@ -0,0 +1,135 @@
+const assert = require('assert');
+const { auth, policies } = require('arsenal');
+const config = require('../config');
+const errors = require('../errors');
+/**
+@class Vault
+* Creates a vault instance for authentication and authorization
+*/
+
+class VaultWrapper extends auth.Vault {
+    create(config) {
+        if (config.vaultd.host) {
+            return new VaultWrapper(config);
+        }
+        return null;
+    }
+
+    constructor(options) {
+        let client;
+        const { host, port } = options.vaultd;
+        const vaultclient = require('vaultclient');
+        if (options.tls) {
+            const { key, cert, ca } = options.tls;
+            client = new vaultclient.Client(host, port, true, key, cert,
+                ca);
+        } else {
+            client = new vaultclient.Client(host, port);
+        }
+        super(client, 'vault');
+    }
+
+    /**
+    * Returns canonical Ids for a given list of account Ids
+    * @param {string[]} accountIds - list of account ids
+    * @param {object} log - Werelogs request logger
+    * @return {Promise} -
+    */
+    getCanonicalIds(accountIds, log) {
+        log.debug('retrieving canonical ids for account ids', {
+            method: 'Vault.getCanonicalIds',
+            accountIds,
+        });
+        return new Promise((resolve, reject) =>
+            this.client.getCanonicalIdsByAccountIds(accountIds,
+                { reqUid: log.getSerializedUids(), logger: log }, (err, res) => {
+                    if (err) {
+                        reject(err);
+                        return;
+                    }
+                    if (!res.message || !res.message.body) {
+                        reject(errors.InternalError);
+                        return;
+                    }
+                    resolve(res.message.body.map(acc => ({
+                        resource: acc.accountId,
+                        id: acc.canonicalId,
+                    })));
+                }));
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    authenticateRequest(request, action, level, resources) {
+        const policyContext = new policies.RequestContext(
+            request.headers,
+            request.query,
+            level,
+            resources,
+            request.ip,
+            request.ctx.encrypted,
+            action,
+            'utapi',
+        );
+
+        return new Promise((resolve, reject) => {
+            auth.server.doAuth(
+                request,
+                request.logger.logger,
+                (err, authInfo, authRes) => {
+                    if (err && err.is && (err.is.InvalidAccessKeyId || err.is.AccessDenied)) {
+                        resolve({ authed: false });
+                        return;
+                    }
+                    if (err) {
+                        reject(err);
+                        return;
+                    }
+
+                    // Only IAM users will return authorizedResources
+                    let authorizedResources = resources;
+                    if (authRes) {
+                        authorizedResources = authRes
+                            .filter(resource => resource.isAllowed)
+                            .map(resource => {
+                                // resource.arn should be of format:
+                                // arn:scality:utapi:::resourcetype/resource
+                                assert(typeof resource.arn === 'string');
+                                assert(resource.arn.indexOf('/') > -1);
+                                return resource.arn.split('/')[1];
+                            });
+                    }
+
+                    resolve({ authed: true, authInfo, authorizedResources });
+                }, 's3', [policyContext],
+            );
+        });
+    }
+
+    getUsersById(userIds, log) {
+        log.debug('retrieving user arns for user ids', {
+            method: 'Vault.getUsersById',
+            userIds,
+        });
+        return new Promise((resolve, reject) =>
+            this.client.getUsersById(userIds,
+                { reqUid: log.getSerializedUids(), logger: log }, (err, res) => {
+                    if (err) {
+                        reject(err);
+                        return;
+                    }
+                    if (!res.message || !res.message.body) {
+                        reject(errors.InternalError);
+                        return;
+                    }
+                    resolve(res.message.body);
+                }));
+    }
+}
+
+const vault = VaultWrapper.create(config);
+auth.setHandler(vault);
+
+module.exports = {
+    VaultWrapper,
+    vault,
+};

libV2/vault/index.js

@@ -0,0 +1,177 @@
+const { vault } = require('./client');
+const metadata = require('../metadata');
+const errors = require('../errors');
+const config = require('../config');
+
+async function authorizeAccountAccessKey(authInfo, level, resources, log) {
+    let authed = false;
+    let authedRes = [];
+
+    log.trace('Authorizing account', { resources });
+
+    switch (level) {
+    // Account keys can only query metrics their own account metrics
+    // So we can short circuit the auth to ->
+    // Did they request their account? Then authorize ONLY their account
+    case 'accounts':
+        authed = resources.some(r => r === authInfo.getShortid());
+        authedRes = authed ? [{ resource: authInfo.getShortid(), id: authInfo.getCanonicalID() }] : [];
+        break;
+
+    // Account keys are allowed access to any of their child users metrics
+    case 'users': {
+        let users;
+        try {
+            users = await vault.getUsersById(resources, log.logger);
+        } catch (error) {
+            log.error('failed to fetch user', { error });
+            throw errors.AccessDenied;
+        }
+        authedRes = users
+            .filter(user => user.parentId === authInfo.getShortid())
+            .map(user => ({ resource: user.id, id: user.id }));
+        authed = authedRes.length !== 0;
+        break;
+    }
+
+    // Accounts are only allowed access if they are the owner of the bucket
+    case 'buckets': {
+        const buckets = await Promise.all(
+            resources.map(async bucket => {
+                try {
+                    const bucketMD = await metadata.getBucket(bucket);
+                    return bucketMD;
+                } catch (error) {
+                    log.error('failed to fetch metadata for bucket', { error, bucket });
+                    throw errors.AccessDenied;
+                }
+            }),
+        );
+
+        authedRes = buckets
+            .filter(bucket => bucket.getOwner() === authInfo.getCanonicalID())
+            .map(bucket => ({ resource: bucket.getName(), id: bucket.getName() }));
+        authed = authedRes.length !== 0;
+        break;
+    }
+
+    // Accounts can not access service resources
+    case 'services':
+        break;
+
+    default:
+        log.error('Unknown metric level', { level });
+        throw new Error(`Unknown metric level ${level}`);
+    }
+
+    return [authed, authedRes];
+}
+
+async function authorizeUserAccessKey(authInfo, level, resources, log) {
+    let authed = false;
+    let authedRes = [];
+
+    log.trace('Authorizing IAM user', { resources });
+
+    // If no resources were authorized by Vault then no further checking is required
+    if (resources.length === 0) {
+        return [false, []];
+    }
+
+    // Get the parent account id from the user's arn
+    const parentAccountId = authInfo.getArn().split(':')[4];
+
+    // All users require an attached policy to query metrics
+    // Additional filtering is performed here to limit access to the user's account
+    switch (level) {
+    // User keys can only query metrics their own account metrics
+    // So we can short circuit the auth to ->
+    // Did they request their account? Then authorize ONLY their account
+    case 'accounts': {
+        authed = resources.some(r => r === parentAccountId);
+        authedRes = authed ? [{ resource: parentAccountId, id: authInfo.getCanonicalID() }] : [];
+        break;
+    }
+
+    // Users can query other user's metrics if they are under the same account
+    case 'users': {
+        let users;
+        try {
+            users = await vault.getUsersById(resources, log.logger);
+        } catch (error) {
+            log.error('failed to fetch user', { error });
+            throw errors.AccessDenied;
+        }
+        authedRes = users
+            .filter(user => user.parentId === parentAccountId)
+            .map(user => ({ resource: user.id, id: user.id }));
+        authed = authedRes.length !== 0;
+        break;
+    }
+
+    // Users can query bucket metrics if they are owned by the same account
+    case 'buckets': {
+        let buckets;
+        try {
+            buckets = await Promise.all(
+                resources.map(bucket => metadata.getBucket(bucket)),
+            );
+        } catch (error) {
+            log.error('failed to fetch metadata for bucket', { error });
+            throw error;
+        }
+        authedRes = buckets
+            .filter(bucket => bucket.getOwner() === authInfo.getCanonicalID())
+            .map(bucket => ({ resource: bucket.getName(), id: bucket.getName() }));
+        authed = authedRes.length !== 0;
+        break;
+    }
+
+    case 'services':
+        break;
+
+    default:
+        log.error('Unknown metric level', { level });
+        throw new Error(`Unknown metric level ${level}`);
+    }
+    return [authed, authedRes];
+}
+
+async function authorizeServiceUser(authInfo, level, resources, log) {
+    log.trace('Authorizing service user', { resources, arn: authInfo.getArn() });
+    // The service user is allowed access to any resource so no checking is done
+    if (level === 'accounts') {
+        const canonicalIds = await vault.getCanonicalIds(resources, log.logger);
+        return [canonicalIds.length !== 0, canonicalIds];
+    }
+
+    return [resources.length !== 0, resources.map(resource => ({ resource, id: resource }))];
+}
+
+
+async function translateAndAuthorize(request, action, level, resources) {
+    const {
+        authed,
+        authInfo,
+        authorizedResources,
+    } = await vault.authenticateRequest(request, action, level, resources);
+
+    if (!authed) {
+        return [false, []];
+    }
+
+    if (config.serviceUser.enabled && authInfo.getArn() === config.serviceUser.arn) {
+        return authorizeServiceUser(authInfo, level, authorizedResources, request.logger);
+    }
+
+    if (authInfo.isRequesterAnIAMUser()) {
+        return authorizeUserAccessKey(authInfo, level, authorizedResources, request.logger);
+    }
+
+    return authorizeAccountAccessKey(authInfo, level, authorizedResources, request.logger);
+}
+
+module.exports = {
+    translateAndAuthorize,
+    vault,
+};

libV2/warp10.js

@@ -1,50 +1,29 @@
 const { Warp10 } = require('@senx/warp10');
+const needle = require('needle');
 const assert = require('assert');
 const { eventFieldsToWarp10, warp10EventType } = require('./constants');
 const _config = require('./config');
 const { LoggerContext } = require('./utils');
-const errors = require('./errors');
 
 const moduleLogger = new LoggerContext({
     module: 'warp10',
 });
 
-
 class Warp10Client {
     constructor(config) {
-        this._writeToken = (config && config.writeToken) || 'writeTokenCI';
-        this._readToken = (config && config.readToken) || 'readTokenCI';
-        this._nodeId = (config && config.nodeId) || _config.nodeId;
+        this._writeToken = (config && config.writeToken) || 'writeTokenStatic';
+        this._readToken = (config && config.readToken) || 'readTokenStatic';
+        this.nodeId = (config && config.nodeId) || _config.nodeId;
         const proto = (config && config.tls) ? 'https' : 'http';
-        const requestTimeout = (config && config.requestTimeout) || 10000;
-        const connectTimeout = (config && config.connectTimeout) || 10000;
-        if (config && config.hosts) {
-            this._clients = config.hosts
-                .map(({ host, port }) => new Warp10(
-                    `${proto}://${host}:${port}`,
-                    requestTimeout,
-                    connectTimeout,
-                ));
-        } else {
-            const host = (config && config.host) || 'localhost';
-            const port = (config && config.port) || 4802;
-            this._clients = [new Warp10(`${proto}://${host}:${port}`, requestTimeout, connectTimeout)];
-        }
+        this._requestTimeout = (config && config.requestTimeout) || 30000;
+        this._connectTimeout = (config && config.connectTimeout) || 30000;
+        const host = (config && config.host) || 'localhost';
+        const port = (config && config.port) || 4802;
+        this._client = new Warp10(`${proto}://${host}:${port}`, this._requestTimeout, this._connectTimeout);
     }
 
-    async _wrapCall(func, params) {
-        // eslint-disable-next-line no-restricted-syntax
-        for (const client of this._clients) {
-            try {
-                // eslint-disable-next-line no-await-in-loop
-                return await func(client, ...params);
-            } catch (error) {
-                moduleLogger.warn('error during warp10 operation, failing over to next host',
-                    { statusCode: error.statusCode, statusMessage: error.statusMessage, error });
-            }
-        }
-        moduleLogger.error('no remaining warp10 hosts to try, unable to complete request');
-        throw errors.InternalError;
+    async update(payload) {
+        return this._client.update(this._writeToken, payload);
     }
 
     static _packEvent(valueType, event) {
@@ -58,12 +37,12 @@ class Warp10Client {
     }
 
     _buildGTSEntry(className, valueType, labels, event) {
-        const _labels = this._clients[0].formatLabels({ node: this._nodeId, ...labels });
+        const _labels = this._client.formatLabels({ node: this.nodeId, ...labels });
         const packed = Warp10Client._packEvent(valueType, event);
         return `${event.timestamp}// ${className}${_labels} ${packed}`;
     }
 
-    async _ingest(warp10, metadata, events) {
+    async ingest(metadata, events) {
         const { className, valueType, labels } = metadata;
         assert.notStrictEqual(className, undefined, 'you must provide a className');
         const payload = events.map(
@@ -74,17 +53,10 @@ class Warp10Client {
                 ev,
             ),
         );
-        const res = await warp10.update(this._writeToken, payload);
+        const res = await this.update(payload);
         return res.count;
     }
 
-    ingest(...params) {
-        return this._wrapCall(
-            this._ingest.bind(this),
-            params,
-        );
-    }
-
     _buildScriptEntry(params) {
         const authInfo = {
             read: this._readToken,
@@ -104,21 +76,15 @@ class Warp10Client {
         return payload.join('\n');
     }
 
-    async _exec(warp10, params) {
+    async exec(params) {
         const payload = this._buildExecPayload(params);
-        const resp = await warp10.exec(payload);
+        const resp = await this._client.exec(payload);
+        moduleLogger.info('warpscript executed', { stats: resp.meta });
         return resp;
     }
 
-    exec(...params) {
-        return this._wrapCall(
-            this._exec.bind(this),
-            params,
-        );
-    }
-
-    async _fetch(warp10, params) {
-        const resp = await warp10.fetch(
+    async fetch(params) {
+        const resp = await this._client.fetch(
             this._readToken,
             params.className,
             params.labels || {},
@@ -130,15 +96,46 @@ class Warp10Client {
         return resp;
     }
 
-    fetch(...params) {
-        return this._wrapCall(
-            this._fetch.bind(this),
-            params,
+    async delete(params) {
+        const {
+            className, labels, start, end,
+        } = params;
+        assert.notStrictEqual(className, undefined, 'you must provide a className');
+        assert.notStrictEqual(start, undefined, 'you must provide a start timestamp');
+        assert.notStrictEqual(end, undefined, 'you must provide a end timestamp');
+        const query = new URLSearchParams([]);
+        query.set('selector', encodeURIComponent(className) + this._client.formatLabels(labels || {}));
+        query.set('start', start.toString());
+        query.set('end', end.toString());
+        const response = await needle(
+            'get',
+            `${this._client.url}/api/v0/delete?${query.toString()}`,
+            {
+                // eslint-disable-next-line camelcase
+                open_timeout: this._connectTimeout,
+                // eslint-disable-next-line camelcase
+                response_timeout: this._requestTimeout,
+                headers: {
+                    'Content-Type': 'text/plain; charset=UTF-8',
+                    'X-Warp10-Token': this._writeToken,
+                },
+            },
         );
+        return { result: response.body };
     }
 }
 
+const clients = _config.warp10.hosts.map(
+    val => new Warp10Client({
+        readToken: _config.warp10.readToken,
+        writeToken: _config.warp10.writeToken,
+        connectTimeout: _config.warp10.connectTimeout,
+        requestTimeout: _config.warp10.requestTimeout,
+        ...val,
+    }),
+);
+
 module.exports = {
     Warp10Client,
-    client: new Warp10Client(_config.warp10),
+    clients,
 };

models/s3metricResponse.json

@@ -46,6 +46,9 @@
         "s3:PutObjectRetention": 0,
         "s3:GetObjectRetention": 0,
         "s3:PutObjectLegalHold": 0,
-        "s3:GetObjectLegalHold": 0
+        "s3:GetObjectLegalHold": 0,
+        "s3:ReplicateObject": 0,
+        "s3:ReplicateTags": 0,
+        "s3:ReplicateDelete": 0
     }
 }

openapi.yaml

@@ -101,6 +101,12 @@ components:
                 type: string
               level:
                 type: string
+    utapi-get-prometheus-metrics:
+      description: metrics to be ingested by prometheus 
+      content:
+        text/plain:
+          schema:
+            type: string
   parameters:
     level:
       in: path
@@ -133,6 +139,16 @@ paths:
           $ref: '#/components/responses/json-error'
         200:
           description: Service is healthy
+  /_/metrics:
+    get:
+      x-router-controller: internal
+      x-iplimit: true
+      operationId: prometheusMetrics
+      responses:
+        default:
+          $ref: '#/components/responses/json-error'
+        200:
+          $ref: '#/components/responses/utapi-get-prometheus-metrics'
   /v2/ingest:
     post:
       x-router-controller: metrics