Feature: Managed Prometheus (AMP) (#5441)

IMPLEMENTATION_COVERAGE.md

@@ -20,6 +20,33 @@
 - [ ] update_certificate_options
 </details>
 
+## amp
+<details>
+<summary>61% implemented</summary>
+
+- [ ] create_alert_manager_definition
+- [ ] create_logging_configuration
+- [X] create_rule_groups_namespace
+- [X] create_workspace
+- [ ] delete_alert_manager_definition
+- [ ] delete_logging_configuration
+- [X] delete_rule_groups_namespace
+- [X] delete_workspace
+- [ ] describe_alert_manager_definition
+- [ ] describe_logging_configuration
+- [X] describe_rule_groups_namespace
+- [X] describe_workspace
+- [X] list_rule_groups_namespaces
+- [X] list_tags_for_resource
+- [X] list_workspaces
+- [ ] put_alert_manager_definition
+- [X] put_rule_groups_namespace
+- [X] tag_resource
+- [X] untag_resource
+- [ ] update_logging_configuration
+- [X] update_workspace_alias
+</details>
+
 ## apigateway
 <details>
 <summary>65% implemented</summary>
@@ -6224,7 +6251,6 @@
 - account
 - acm-pca
 - alexaforbusiness
-- amp
 - amplify
 - amplifybackend
 - amplifyuibuilder

docs/docs/services/amp.rst

@@ -0,0 +1,75 @@
+.. _implementedservice_amp:
+
+.. |start-h3| raw:: html
+
+    <h3>
+
+.. |end-h3| raw:: html
+
+    </h3>
+
+===
+amp
+===
+
+.. autoclass:: moto.amp.models.PrometheusServiceBackend
+
+|start-h3| Example usage |end-h3|
+
+.. sourcecode:: python
+
+            @mock_amp
+            def test_amp_behaviour:
+                boto3.client("amp")
+                ...
+
+
+
+|start-h3| Implemented features for this service |end-h3|
+
+- [ ] create_alert_manager_definition
+- [ ] create_logging_configuration
+- [X] create_rule_groups_namespace
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+- [X] create_workspace
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+- [ ] delete_alert_manager_definition
+- [ ] delete_logging_configuration
+- [X] delete_rule_groups_namespace
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+- [X] delete_workspace
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+- [ ] describe_alert_manager_definition
+- [ ] describe_logging_configuration
+- [X] describe_rule_groups_namespace
+- [X] describe_workspace
+- [X] list_rule_groups_namespaces
+- [X] list_tags_for_resource
+- [X] list_workspaces
+- [ ] put_alert_manager_definition
+- [X] put_rule_groups_namespace
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+- [X] tag_resource
+- [X] untag_resource
+- [ ] update_logging_configuration
+- [X] update_workspace_alias
+  
+        The ClientToken-parameter is not yet implemented
+        
+
+

moto/__init__.py

@@ -16,6 +16,7 @@ def lazy_load(module_name, element, boto3_name=None, backend=None):
 
 
 mock_acm = lazy_load(".acm", "mock_acm")
+mock_amp = lazy_load(".amp", "mock_amp")
 mock_apigateway = lazy_load(".apigateway", "mock_apigateway")
 mock_apigatewayv2 = lazy_load(".apigatewayv2", "mock_apigatewayv2")
 mock_appsync = lazy_load(".appsync", "mock_appsync")

moto/amp/__init__.py

@@ -0,0 +1,5 @@
+"""amp module initialization; sets value for base decorator."""
+from .models import amp_backends
+from ..core.models import base_decorator
+
+mock_amp = base_decorator(amp_backends)

moto/amp/exceptions.py

@@ -0,0 +1,40 @@
+import json
+from moto.core.exceptions import JsonRESTError
+
+
+class AmpException(JsonRESTError):
+    pass
+
+
+class ResourceNotFoundException(AmpException):
+    def __init__(self, message, resource_id, resource_type):
+        super().__init__("ResourceNotFoundException", message)
+        self.description = json.dumps(
+            {
+                "resourceId": resource_id,
+                "message": self.message,
+                "resourceType": resource_type,
+            }
+        )
+
+
+class WorkspaceNotFound(ResourceNotFoundException):
+    code = 404
+
+    def __init__(self, workspace_id):
+        super().__init__(
+            "Workspace not found",
+            resource_id=workspace_id,
+            resource_type="AWS::APS::Workspace",
+        )
+
+
+class RuleGroupNamespaceNotFound(ResourceNotFoundException):
+    code = 404
+
+    def __init__(self, name):
+        super().__init__(
+            "RuleGroupNamespace not found",
+            resource_id=name,
+            resource_type="AWS::APS::RuleGroupNamespace",
+        )

moto/amp/models.py

@@ -0,0 +1,168 @@
+"""PrometheusServiceBackend class with methods for supported APIs."""
+
+from moto.core import BaseBackend, BaseModel
+from moto.core.utils import BackendDict, unix_time
+from moto.utilities.paginator import paginate
+from moto.utilities.tagging_service import TaggingService
+from typing import Dict
+from uuid import uuid4
+from .exceptions import RuleGroupNamespaceNotFound, WorkspaceNotFound
+from .utils import PAGINATION_MODEL
+
+
+class RuleGroupNamespace(BaseModel):
+    def __init__(self, account_id, region, workspace_id, name, data, tag_fn):
+        self.name = name
+        self.data = data
+        self.tag_fn = tag_fn
+        self.arn = f"arn:aws:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{self.name}"
+        self.created_at = unix_time()
+        self.modified_at = self.created_at
+
+    def update(self, new_data):
+        self.data = new_data
+        self.modified_at = unix_time()
+
+    def to_dict(self):
+        return {
+            "name": self.name,
+            "arn": self.arn,
+            "status": {"statusCode": "ACTIVE"},
+            "createdAt": self.created_at,
+            "modifiedAt": self.modified_at,
+            "data": self.data,
+            "tags": self.tag_fn(self.arn),
+        }
+
+
+class Workspace(BaseModel):
+    def __init__(self, account_id, region, alias, tag_fn):
+        self.alias = alias
+        self.workspace_id = f"ws-{uuid4()}"
+        self.arn = f"arn:aws:aps:{region}:{account_id}:workspace/{self.workspace_id}"
+        self.endpoint = f"https://aps-workspaces.{region}.amazonaws.com/workspaces/{self.workspace_id}/"
+        self.status = {"statusCode": "ACTIVE"}
+        self.created_at = unix_time()
+        self.tag_fn = tag_fn
+        self.rule_group_namespaces = dict()
+
+    def to_dict(self):
+        return {
+            "alias": self.alias,
+            "arn": self.arn,
+            "workspaceId": self.workspace_id,
+            "status": self.status,
+            "createdAt": self.created_at,
+            "prometheusEndpoint": self.endpoint,
+            "tags": self.tag_fn(self.arn),
+        }
+
+
+class PrometheusServiceBackend(BaseBackend):
+    """Implementation of PrometheusService APIs."""
+
+    def __init__(self, region_name, account_id):
+        super().__init__(region_name, account_id)
+        self.workspaces: Dict(str, Workspace) = dict()
+        self.tagger = TaggingService()
+
+    def create_workspace(self, alias, tags):
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        workspace = Workspace(
+            self.account_id,
+            self.region_name,
+            alias=alias,
+            tag_fn=self.list_tags_for_resource,
+        )
+        self.workspaces[workspace.workspace_id] = workspace
+        self.tag_resource(workspace.arn, tags)
+        return workspace
+
+    def describe_workspace(self, workspace_id) -> Workspace:
+        if workspace_id not in self.workspaces:
+            raise WorkspaceNotFound(workspace_id)
+        return self.workspaces[workspace_id]
+
+    def list_tags_for_resource(self, resource_arn):
+        return self.tagger.get_tag_dict_for_resource(resource_arn)
+
+    def update_workspace_alias(self, alias, workspace_id):
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        self.workspaces[workspace_id].alias = alias
+
+    def delete_workspace(self, workspace_id):
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        self.workspaces.pop(workspace_id, None)
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_workspaces(self, alias):
+        if alias:
+            return [w for w in self.workspaces.values() if w.alias == alias]
+        return list(self.workspaces.values())
+
+    def tag_resource(self, resource_arn, tags):
+        tags = self.tagger.convert_dict_to_tags_input(tags)
+        self.tagger.tag_resource(resource_arn, tags)
+
+    def untag_resource(self, resource_arn, tag_keys):
+        self.tagger.untag_resource_using_names(resource_arn, tag_keys)
+
+    def create_rule_groups_namespace(
+        self, data, name, tags, workspace_id
+    ) -> RuleGroupNamespace:
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        workspace = self.describe_workspace(workspace_id)
+        group = RuleGroupNamespace(
+            account_id=self.account_id,
+            region=self.region_name,
+            workspace_id=workspace_id,
+            name=name,
+            data=data,
+            tag_fn=self.list_tags_for_resource,
+        )
+        workspace.rule_group_namespaces[name] = group
+        self.tag_resource(group.arn, tags)
+        return group
+
+    def delete_rule_groups_namespace(self, name, workspace_id) -> None:
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        ws = self.describe_workspace(workspace_id)
+        ws.rule_group_namespaces.pop(name, None)
+
+    def describe_rule_groups_namespace(self, name, workspace_id) -> RuleGroupNamespace:
+        ws = self.describe_workspace(workspace_id)
+        if name not in ws.rule_group_namespaces:
+            raise RuleGroupNamespaceNotFound(name=name)
+        return ws.rule_group_namespaces[name]
+
+    def put_rule_groups_namespace(self, data, name, workspace_id) -> RuleGroupNamespace:
+        """
+        The ClientToken-parameter is not yet implemented
+        """
+        ns = self.describe_rule_groups_namespace(name=name, workspace_id=workspace_id)
+        ns.update(data)
+        return ns
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_rule_groups_namespaces(self, name, workspace_id):
+        ws = self.describe_workspace(workspace_id)
+        if name:
+            return [
+                ns
+                for ns_name, ns in ws.rule_group_namespaces.items()
+                if ns_name.startswith(name)
+            ]
+        return list(ws.rule_group_namespaces.values())
+
+
+amp_backends = BackendDict(PrometheusServiceBackend, "amp")

moto/amp/responses.py

@@ -0,0 +1,141 @@
+"""Handles incoming amp requests, invokes methods, returns responses."""
+import json
+
+from moto.core.responses import BaseResponse
+from .models import amp_backends, PrometheusServiceBackend
+from urllib.parse import unquote
+
+
+class PrometheusServiceResponse(BaseResponse):
+    """Handler for PrometheusService requests and responses."""
+
+    def tags(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+        if request.method == "GET":
+            return self.list_tags_for_resource()
+        if request.method == "POST":
+            return self.tag_resource()
+        if request.method == "DELETE":
+            return self.untag_resource()
+
+    def __init__(self):
+        super().__init__(service_name="amp")
+
+    @property
+    def amp_backend(self) -> PrometheusServiceBackend:
+        """Return backend instance specific for this region."""
+        return amp_backends[self.current_account][self.region]
+
+    def create_workspace(self):
+        params = json.loads(self.body)
+        alias = params.get("alias")
+        tags = params.get("tags")
+        workspace = self.amp_backend.create_workspace(alias=alias, tags=tags)
+        return json.dumps(dict(workspace.to_dict()))
+
+    def describe_workspace(self):
+        workspace_id = self.path.split("/")[-1]
+        workspace = self.amp_backend.describe_workspace(workspace_id=workspace_id)
+        return json.dumps(dict(workspace=workspace.to_dict()))
+
+    def list_tags_for_resource(self):
+        resource_arn = unquote(self.path).split("tags/")[-1]
+        tags = self.amp_backend.list_tags_for_resource(resource_arn=resource_arn)
+        return json.dumps(dict(tags=tags))
+
+    def update_workspace_alias(self):
+        params = json.loads(self.body)
+        alias = params.get("alias")
+        workspace_id = self.path.split("/")[-2]
+        self.amp_backend.update_workspace_alias(alias=alias, workspace_id=workspace_id)
+        return json.dumps(dict())
+
+    def delete_workspace(self):
+        workspace_id = self.path.split("/")[-1]
+        self.amp_backend.delete_workspace(workspace_id=workspace_id)
+        return json.dumps(dict())
+
+    def list_workspaces(self):
+        alias = self._get_param("alias")
+        max_results = self._get_int_param("maxResults")
+        next_token = self._get_param("nextToken")
+        workspaces, next_token = self.amp_backend.list_workspaces(
+            alias, max_results=max_results, next_token=next_token
+        )
+        return json.dumps(
+            {"nextToken": next_token, "workspaces": [w.to_dict() for w in workspaces]}
+        )
+
+    def tag_resource(self):
+        params = json.loads(self.body)
+        resource_arn = unquote(self.path).split("tags/")[-1]
+        tags = params.get("tags")
+        self.amp_backend.tag_resource(resource_arn=resource_arn, tags=tags)
+        return json.dumps(dict())
+
+    def untag_resource(self):
+        resource_arn = unquote(self.path).split("tags/")[-1]
+        tag_keys = self.querystring.get("tagKeys", [])
+        self.amp_backend.untag_resource(resource_arn=resource_arn, tag_keys=tag_keys)
+        return json.dumps(dict())
+
+    def create_rule_groups_namespace(self):
+        params = json.loads(self.body)
+        data = params.get("data")
+        name = params.get("name")
+        tags = params.get("tags")
+        workspace_id = unquote(self.path).split("/")[-2]
+        rule_group_namespace = self.amp_backend.create_rule_groups_namespace(
+            data=data,
+            name=name,
+            tags=tags,
+            workspace_id=workspace_id,
+        )
+        return json.dumps(rule_group_namespace.to_dict())
+
+    def delete_rule_groups_namespace(self):
+        name = unquote(self.path).split("/")[-1]
+        workspace_id = unquote(self.path).split("/")[-3]
+        self.amp_backend.delete_rule_groups_namespace(
+            name=name,
+            workspace_id=workspace_id,
+        )
+        return json.dumps(dict())
+
+    def describe_rule_groups_namespace(self):
+        name = unquote(self.path).split("/")[-1]
+        workspace_id = unquote(self.path).split("/")[-3]
+        ns = self.amp_backend.describe_rule_groups_namespace(
+            name=name, workspace_id=workspace_id
+        )
+        return json.dumps(dict(ruleGroupsNamespace=ns.to_dict()))
+
+    def put_rule_groups_namespace(self):
+        params = json.loads(self.body)
+        data = params.get("data")
+        name = unquote(self.path).split("/")[-1]
+        workspace_id = unquote(self.path).split("/")[-3]
+        ns = self.amp_backend.put_rule_groups_namespace(
+            data=data,
+            name=name,
+            workspace_id=workspace_id,
+        )
+        return json.dumps(ns.to_dict())
+
+    def list_rule_groups_namespaces(self):
+        max_results = self._get_int_param("maxResults")
+        next_token = self._get_param("nextToken")
+        name = self._get_param("name")
+        workspace_id = unquote(self.path).split("/")[-2]
+        namespaces, next_token = self.amp_backend.list_rule_groups_namespaces(
+            max_results=max_results,
+            name=name,
+            next_token=next_token,
+            workspace_id=workspace_id,
+        )
+        return json.dumps(
+            dict(
+                nextToken=next_token,
+                ruleGroupsNamespaces=[ns.to_dict() for ns in namespaces],
+            )
+        )

moto/amp/urls.py

@@ -0,0 +1,21 @@
+"""amp base URL and path."""
+from .responses import PrometheusServiceResponse
+
+url_bases = [
+    r"https?://aps\.(.+)\.amazonaws\.com",
+]
+
+
+response = PrometheusServiceResponse()
+
+
+url_paths = {
+    "{0}/workspaces$": response.dispatch,
+    "{0}/workspaces/(?P<workspace_id>[^/]+)$": response.dispatch,
+    "{0}/workspaces/(?P<workspace_id>[^/]+)/alias$": response.dispatch,
+    "{0}/workspaces/(?P<workspace_id>[^/]+)/rulegroupsnamespaces$": response.dispatch,
+    "{0}/workspaces/(?P<workspace_id>[^/]+)/rulegroupsnamespaces/(?P<name>[^/]+)$": response.dispatch,
+    "{0}/tags/(?P<resource_arn>[^/]+)$": response.dispatch,
+    "{0}/tags/(?P<arn_prefix>[^/]+)/(?P<workspace_id>[^/]+)$": response.tags,
+    "{0}/tags/(?P<arn_prefix>[^/]+)/(?P<workspace_id>[^/]+)/(?P<ns_name>[^/]+)$": response.tags,
+}

moto/amp/utils.py

@@ -0,0 +1,14 @@
+PAGINATION_MODEL = {
+    "list_workspaces": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,  # This should be the sum of the directory limits
+        "unique_attribute": "arn",
+    },
+    "list_rule_groups_namespaces": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,  # This should be the sum of the directory limits
+        "unique_attribute": "name",
+    },
+}

moto/backend_index.py

@@ -3,6 +3,7 @@ import re
 
 backend_url_patterns = [
     ("acm", re.compile("https?://acm\\.(.+)\\.amazonaws\\.com")),
+    ("amp", re.compile("https?://aps\\.(.+)\\.amazonaws\\.com")),
     ("apigateway", re.compile("https?://apigateway\\.(.+)\\.amazonaws.com")),
     (
         "applicationautoscaling",

tests/terraformtests/terraform-tests.success.txt

@@ -1,5 +1,8 @@
 acm:
   - TestAccACMCertificateDataSource
+amp:
+  - TestAccAMPWorkspace
+  - TestAccAMPRuleGroupNamespace
 apigateway:
   - TestAccAPIGatewayGatewayResponse
 apigatewayv2:

tests/test_amp/test_amp_rulegroupnamespaces.py

@@ -0,0 +1,173 @@
+"""Unit tests for amp-supported APIs."""
+import boto3
+import pytest
+import sure  # noqa # pylint: disable=unused-import
+
+from botocore.exceptions import ClientError
+from moto import mock_amp
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_amp
+def test_create_rule_groups_namespace():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    workspace_id = client.create_workspace()["workspaceId"]
+    resp = client.create_rule_groups_namespace(
+        data=b"asdf", name="my first rule group", workspaceId=workspace_id
+    )
+
+    resp.should.have.key("arn")
+    resp.should.have.key("name").equals("my first rule group")
+    resp.should.have.key("status")
+
+
+@mock_amp
+def test_delete_rule_groups_namespace():
+    client = boto3.client("amp", region_name="us-east-2")
+    workspace_id = client.create_workspace()["workspaceId"]
+    client.create_rule_groups_namespace(
+        data=b"asdf", name="myname", workspaceId=workspace_id
+    )
+
+    client.delete_rule_groups_namespace(name="myname", workspaceId=workspace_id)
+
+    with pytest.raises(ClientError) as exc:
+        client.describe_rule_groups_namespace(name="myname", workspaceId=workspace_id)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ResourceNotFoundException")
+    err["Message"].should.equal("RuleGroupNamespace not found")
+
+
+@mock_amp
+def test_describe_rule_groups_namespace():
+    client = boto3.client("amp", region_name="us-east-2")
+
+    workspace_id = client.create_workspace()["workspaceId"]
+    client.create_rule_groups_namespace(
+        data=b"asdf", name="myname", workspaceId=workspace_id
+    )
+
+    resp = client.describe_rule_groups_namespace(
+        name="myname", workspaceId=workspace_id
+    )
+    resp.should.have.key("ruleGroupsNamespace")
+    ns = resp["ruleGroupsNamespace"]
+
+    ns.should.have.key("arn")
+    ns.should.have.key("createdAt")
+    ns.should.have.key("data").equals(b"asdf")
+    ns.should.have.key("modifiedAt")
+    ns.should.have.key("name").equals("myname")
+    ns.should.have.key("status")
+
+
+@mock_amp
+def test_put_rule_groups_namespace():
+    client = boto3.client("amp", region_name="eu-west-1")
+
+    workspace_id = client.create_workspace()["workspaceId"]
+    client.create_rule_groups_namespace(
+        data=b"asdf", name="myname", workspaceId=workspace_id
+    )
+
+    client.put_rule_groups_namespace(
+        name="myname", workspaceId=workspace_id, data=b"updated"
+    )
+
+    resp = client.describe_rule_groups_namespace(
+        name="myname", workspaceId=workspace_id
+    )
+    resp.should.have.key("ruleGroupsNamespace")
+    ns = resp["ruleGroupsNamespace"]
+
+    ns.should.have.key("arn")
+    ns.should.have.key("createdAt")
+    ns.should.have.key("data").equals(b"updated")
+
+
+@mock_amp
+def test_list_rule_groups_namespaces():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    w_id = client.create_workspace()["workspaceId"]
+    for idx in range(15):
+        client.create_rule_groups_namespace(
+            data=b"a", name=f"ns{idx}", workspaceId=w_id
+        )
+
+    resp = client.list_rule_groups_namespaces(workspaceId=w_id)
+    resp.should.have.key("ruleGroupsNamespaces").length_of(15)
+    resp.shouldnt.have.key("nextToken")
+
+    resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns1")
+    resp.should.have.key("ruleGroupsNamespaces").length_of(6)
+    names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]]
+    set(names).should.equal({"ns10", "ns13", "ns1", "ns12", "ns11", "ns14"})
+
+    resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns10")
+    resp.should.have.key("ruleGroupsNamespaces").length_of(1)
+    names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]]
+    set(names).should.equal({"ns10"})
+
+
+@mock_amp
+def test_list_rule_groups_namespaces__paginated():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    w_id = client.create_workspace()["workspaceId"]
+    for idx in range(125):
+        client.create_rule_groups_namespace(
+            data=b"a", name=f"ns{idx}", workspaceId=w_id
+        )
+
+    # default pagesize is 100
+    page1 = client.list_rule_groups_namespaces(workspaceId=w_id)
+    page1.should.have.key("ruleGroupsNamespaces").length_of(100)
+    page1.should.have.key("nextToken")
+
+    # We can ask for a smaller pagesize
+    page2 = client.list_rule_groups_namespaces(
+        workspaceId=w_id, maxResults=15, nextToken=page1["nextToken"]
+    )
+    page2.should.have.key("ruleGroupsNamespaces").length_of(15)
+    page2.should.have.key("nextToken")
+
+    page3 = client.list_rule_groups_namespaces(
+        workspaceId=w_id, maxResults=15, nextToken=page2["nextToken"]
+    )
+    page3.should.have.key("ruleGroupsNamespaces").length_of(10)
+    page3.shouldnt.have.key("nextToken")
+
+    # We could request all of them in one go
+    full_page = client.list_rule_groups_namespaces(workspaceId=w_id, maxResults=150)
+    full_page.should.have.key("ruleGroupsNamespaces").length_of(125)
+    full_page.shouldnt.have.key("nextToken")
+
+
+@mock_amp
+def test_tag_resource():
+    client = boto3.client("amp", region_name="us-east-2")
+
+    w_id = client.create_workspace()["workspaceId"]
+    ns = client.create_rule_groups_namespace(
+        data=b"a", name="ns", workspaceId=w_id, tags={"t": "v"}
+    )
+
+    arn = ns["arn"]
+
+    client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"})
+
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal(
+        {"t": "v", "t1": "v1", "t2": "v2"}
+    )
+    client.describe_rule_groups_namespace(workspaceId=w_id, name="ns")[
+        "ruleGroupsNamespace"
+    ]["tags"].should.equal({"t": "v", "t1": "v1", "t2": "v2"})
+
+    client.untag_resource(resourceArn=arn, tagKeys=["t1"])
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal(
+        {"t": "v", "t2": "v2"}
+    )
+
+    client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"])
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({})

tests/test_amp/test_amp_workspaces.py

@@ -0,0 +1,149 @@
+import boto3
+import pytest
+import sure  # noqa # pylint: disable=unused-import
+
+from botocore.exceptions import ClientError
+from moto import mock_amp
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_amp
+def test_create_workspace():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    resp = client.create_workspace(alias="test", clientToken="mytoken")
+
+    resp.should.have.key("arn")
+    resp.should.have.key("status").equals({"statusCode": "ACTIVE"})
+    resp.should.have.key("workspaceId")
+
+
+@mock_amp
+def test_describe_workspace():
+    client = boto3.client("amp", region_name="eu-west-1")
+    workspace_id = client.create_workspace(alias="test", clientToken="mytoken")[
+        "workspaceId"
+    ]
+
+    resp = client.describe_workspace(workspaceId=workspace_id)
+    resp.should.have.key("workspace")
+
+    workspace = resp["workspace"]
+    workspace.should.have.key("alias")
+    workspace.should.have.key("arn")
+    workspace.should.have.key("createdAt")
+    workspace.should.have.key("prometheusEndpoint")
+    workspace.should.have.key("status").equals({"statusCode": "ACTIVE"})
+    workspace.should.have.key("workspaceId").equals(workspace_id)
+
+
+@mock_amp
+def test_list_workspaces():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    client.create_workspace(alias="test")
+    client.create_workspace(alias="another")
+    client.create_workspace()
+
+    resp = client.list_workspaces()
+    resp.should.have.key("workspaces").length_of(3)
+    resp.shouldnt.have.key("nextToken")
+
+    resp = client.list_workspaces(alias="another")
+    resp.should.have.key("workspaces").length_of(1)
+    resp["workspaces"][0].should.have.key("alias").equals("another")
+
+
+@mock_amp
+def test_list_workspaces__paginated():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    for _ in range(125):
+        client.create_workspace()
+
+    # default pagesize is 100
+    page1 = client.list_workspaces()
+    page1.should.have.key("workspaces").length_of(100)
+    page1.should.have.key("nextToken")
+
+    # We can ask for a smaller pagesize
+    page2 = client.list_workspaces(maxResults=15, nextToken=page1["nextToken"])
+    page2.should.have.key("workspaces").length_of(15)
+    page2.should.have.key("nextToken")
+
+    page3 = client.list_workspaces(maxResults=15, nextToken=page2["nextToken"])
+    page3.should.have.key("workspaces").length_of(10)
+    page3.shouldnt.have.key("nextToken")
+
+    # We could request all of them in one go
+    full_page = client.list_workspaces(maxResults=150)
+    full_page.should.have.key("workspaces").length_of(125)
+    full_page.shouldnt.have.key("nextToken")
+
+
+@mock_amp
+def test_list_tags_for_resource():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+    arn = client.create_workspace(
+        alias="test", clientToken="mytoken", tags={"t1": "v1", "t2": "v2"}
+    )["arn"]
+
+    resp = client.list_tags_for_resource(resourceArn=arn)
+    resp.should.have.key("tags").equals({"t1": "v1", "t2": "v2"})
+
+
+@mock_amp
+def test_update_workspace_alias():
+    client = boto3.client("amp", region_name="ap-southeast-1")
+
+    workspace_id = client.create_workspace(alias="initial")["workspaceId"]
+
+    w = client.describe_workspace(workspaceId=workspace_id)["workspace"]
+    w.should.have.key("alias").equals("initial")
+
+    client.update_workspace_alias(alias="updated", workspaceId=workspace_id)
+
+    w = client.describe_workspace(workspaceId=workspace_id)["workspace"]
+    w.should.have.key("alias").equals("updated")
+
+
+@mock_amp
+def test_delete_workspace():
+    client = boto3.client("amp", region_name="us-east-2")
+
+    workspace_id = client.create_workspace(alias="test", clientToken="mytoken")[
+        "workspaceId"
+    ]
+
+    client.delete_workspace(workspaceId=workspace_id)
+
+    with pytest.raises(ClientError) as exc:
+        client.describe_workspace(workspaceId=workspace_id)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ResourceNotFoundException")
+    err["Message"].should.equal("Workspace not found")
+
+
+@mock_amp
+def test_tag_resource():
+    client = boto3.client("amp", region_name="us-east-2")
+
+    workspace = client.create_workspace(alias="test", tags={"t": "v"})
+    arn = workspace["arn"]
+    workspace_id = workspace["workspaceId"]
+
+    client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"})
+
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal(
+        {"t": "v", "t1": "v1", "t2": "v2"}
+    )
+    client.describe_workspace(workspaceId=workspace_id)["workspace"][
+        "tags"
+    ].should.equal({"t": "v", "t1": "v1", "t2": "v2"})
+
+    client.untag_resource(resourceArn=arn, tagKeys=["t1"])
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal(
+        {"t": "v", "t2": "v2"}
+    )
+
+    client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"])
+    client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({})