diff --git a/README.md b/README.md
index 91e20009..c5601921 100644
--- a/README.md
+++ b/README.md
@@ -244,6 +244,8 @@ deployed then see file `actions.yaml`.
 * `osd-in`
 * `osd-out`
 * `security-checklist`
+* `start`
+* `stop`
 * `zap-disk`
 
 ## Working with OSDs
@@ -293,6 +295,25 @@ Example:
 
     juju run-action --wait ceph-osd/4 osd-in
 
+### Managing ceph OSDs
+
+Use the `stop` and `start` actions to manage ceph OSD services within the unit.
+Both actions take one parameter, `osds`, which should contain comma-separated
+numerical IDs of `ceph-osd` services or the keyword `all`.
+
+Example:
+
+    # stop ceph-osd@0 and ceph-osd@1
+    juju run-action --wait ceph-osd/0 stop osds=0,1
+    # start all ceph-osd services on the unit
+    juju run-action --wait ceph-osd/0 start osds=all
+
+ > **Note**: Stopping ceph-osd services will put the unit into the blocked
+ state.
+ 
+ > **Important**: This action is not available on Trusty due to reliance on
+ systemd.
+
 ## Working with disks
 
 ### List disks
diff --git a/actions.yaml b/actions.yaml
index 880a922d..1674a08d 100644
--- a/actions.yaml
+++ b/actions.yaml
@@ -84,5 +84,25 @@ zap-disk:
   required:
     - devices
     - i-really-mean-it
+start:
+  description: |
+    \
+        Start OSD by ID
+        Documentation: https://jaas.ai/ceph-osd/
+  params:
+    osds:
+      description: A comma-separated list of OSD IDs to start (or keyword 'all')
+  required:
+    - osds
+stop:
+  description: |
+    \
+        Stop OSD by ID
+        Documentation: https://jaas.ai/ceph-osd/
+  params:
+    osds:
+      description: A comma-separated list of OSD IDs to stop (or keyword 'all')
+  required:
+    - osds
 security-checklist:
   description: Validate the running configuration against the OpenStack security guides checklist
diff --git a/actions/service.py b/actions/service.py
new file mode 100755
index 00000000..1b8fe1ff
--- /dev/null
+++ b/actions/service.py
@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+#
+# Copyright 2020 Canonical Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import os
+import sys
+import shutil
+import subprocess
+
+
+sys.path.append('lib')
+sys.path.append('hooks')
+
+from charmhelpers.core.hookenv import (
+    function_fail,
+    function_get,
+    log,
+    WARNING,
+)
+from ceph_hooks import assess_status
+
+START = 'start'
+STOP = 'stop'
+
+ALL = 'all'
+
+
+def systemctl_execute(action, services):
+    """
+    Execute `systemctl` action on specified services.
+
+    Action can be either 'start' or 'stop' (defined by global constants
+    START, STOP). Parameter `services` is list of service names on which the
+    action will be executed. If the parameter `services` contains constant
+    ALL, the action will be executed on all ceph-osd services.
+
+    :param action: Action to be executed (start or stop)
+    :type action: str
+    :param services: List of services to be targetd by the action
+    :type services: list[str]
+    :return: None
+    """
+    if ALL in services:
+        cmd = ['systemctl', action, 'ceph-osd.target']
+    else:
+        cmd = ['systemctl', action] + services
+    subprocess.check_call(cmd, timeout=300)
+
+
+def osd_ids_to_service_names(osd_ids):
+    """
+    Transform set of OSD IDs into the list of respective service names.
+
+    Example:
+        >>> osd_ids_to_service_names({0,1})
+        ['ceph-osd@0.service', 'ceph-osd@1.service']
+
+    :param osd_ids: Set of service IDs to be converted
+    :type osd_ids: set[str | int]
+    :return: List of service names
+    :rtype: list[str]
+    """
+    service_list = []
+    for id_ in osd_ids:
+        if id_ == ALL:
+            service_list.append(ALL)
+        else:
+            service_list.append("ceph-osd@{}.service".format(id_))
+    return service_list
+
+
+def check_service_is_present(service_list):
+    """
+    Checks that every service, from the `service_list` parameter exists
+    on the system. Raises RuntimeError if any service is missing.
+
+    :param service_list: List of systemd services
+    :type service_list: list[str]
+    :raises RuntimeError: if any service is missing
+    """
+    if ALL in service_list:
+        return
+
+    service_list_cmd = ['systemctl', 'list-units', '--full',
+                        '--all', '--no-pager', '-t', 'service']
+    present_services = subprocess.run(service_list_cmd,
+                                      stdout=subprocess.PIPE,
+                                      timeout=30).stdout.decode('utf-8')
+
+    missing_services = []
+    for service_name in service_list:
+        if service_name not in present_services:
+            missing_services.append(service_name)
+
+    if missing_services:
+        raise RuntimeError('Some services are not present on this '
+                           'unit: {}'.format(missing_services))
+
+
+def parse_arguments():
+    """
+    Fetch action arguments and parse them from comma separated list to
+    the set of OSD IDs
+
+    :return: Set of OSD IDs
+    :rtype: set(str)
+    """
+    raw_arg = function_get('osds')
+
+    if raw_arg is None:
+        raise RuntimeError('Action argument "osds" is missing')
+    args = set()
+
+    # convert OSD IDs from user's input into the set
+    for osd_id in str(raw_arg).split(','):
+        args.add(osd_id.strip())
+
+    if ALL in args and len(args) != 1:
+        args = {ALL}
+        log('keyword "all" was found in "osds" argument. Dropping other '
+            'explicitly defined OSD IDs', WARNING)
+
+    return args
+
+
+def execute_action(action):
+    """Core implementation of the 'start'/'stop' actions
+
+    :param action: Either START or STOP (see global constants)
+    :return: None
+    """
+    if action not in (START, STOP):
+        raise RuntimeError('Unknown action "{}"'.format(action))
+
+    osds = parse_arguments()
+    services = osd_ids_to_service_names(osds)
+
+    check_service_is_present(services)
+
+    systemctl_execute(action, services)
+
+    assess_status()
+
+
+def stop():
+    """Shortcut to execute 'stop' action"""
+    execute_action(STOP)
+
+
+def start():
+    """Shortcut to execute 'start' action"""
+    execute_action(START)
+
+
+ACTIONS = {'stop': stop,
+           'start': start,
+           }
+
+
+def main(args):
+    action_name = os.path.basename(args.pop(0))
+    try:
+        action = ACTIONS[action_name]
+    except KeyError:
+        s = "Action {} undefined".format(action_name)
+        function_fail(s)
+        return
+    else:
+        try:
+            log("Running action '{}'.".format(action_name))
+            if shutil.which('systemctl') is None:
+                raise RuntimeError("This action requires systemd")
+            action()
+        except Exception as e:
+            function_fail("Action '{}' failed: {}".format(action_name, str(e)))
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))
diff --git a/actions/start b/actions/start
new file mode 120000
index 00000000..12afe70c
--- /dev/null
+++ b/actions/start
@@ -0,0 +1 @@
+service.py
\ No newline at end of file
diff --git a/actions/stop b/actions/stop
new file mode 120000
index 00000000..12afe70c
--- /dev/null
+++ b/actions/stop
@@ -0,0 +1 @@
+service.py
\ No newline at end of file
diff --git a/tests/tests.yaml b/tests/tests.yaml
index 1f6354e8..02c1a6bc 100644
--- a/tests/tests.yaml
+++ b/tests/tests.yaml
@@ -22,6 +22,7 @@ tests:
   - zaza.openstack.charm_tests.ceph.tests.CephRelationTest
   - zaza.openstack.charm_tests.ceph.tests.CephTest
   - zaza.openstack.charm_tests.ceph.osd.tests.SecurityTest
+  - zaza.openstack.charm_tests.ceph.osd.tests.ServiceTest
 tests_options:
   force_deploy:
     - groovy-victoria
diff --git a/unit_tests/test_actions_service.py b/unit_tests/test_actions_service.py
new file mode 100644
index 00000000..daa64e8d
--- /dev/null
+++ b/unit_tests/test_actions_service.py
@@ -0,0 +1,223 @@
+# Copyright 2020 Canonical Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import mock
+from contextlib import contextmanager
+from copy import copy
+
+from actions import service
+
+from test_utils import CharmTestCase
+
+
+class CompletedProcessMock:
+    def __init__(self, stdout=b'', stderr=b''):
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class ServiceActionTests(CharmTestCase):
+    _PRESENT_SERVICES = [
+        "ceph-osd@0.service",
+        "ceph-osd@1.service",
+        "ceph-osd@2.service",
+    ]
+
+    _TARGET_ALL = 'ceph-osd.target'
+
+    _CHECK_CALL_TIMEOUT = 300
+
+    def __init__(self, methodName='runTest'):
+        super(ServiceActionTests, self).__init__(methodName)
+        self._func_args = {'osds': None}
+
+    def setUp(self, obj=None, patches=None):
+        super(ServiceActionTests, self).setUp(
+            service,
+            ['subprocess', 'function_fail', 'function_get',
+             'log', 'assess_status', 'shutil']
+        )
+        present_services = '\n'.join(self._PRESENT_SERVICES).encode('utf-8')
+
+        self.shutil.which.return_value = '/bin/systemctl'
+        self.subprocess.check_call.return_value = None
+        self.function_get.side_effect = self.function_get_side_effect
+        self.subprocess.run.return_value = CompletedProcessMock(
+            stdout=present_services)
+
+    def function_get_side_effect(self, arg):
+        return self._func_args.get(arg)
+
+    @contextmanager
+    def func_call_arguments(self, osds=None):
+        default = copy(self._func_args)
+        try:
+            self._func_args = {'osds': osds}
+            yield
+        finally:
+            self._func_args = copy(default)
+
+    def assert_action_start_fail(self, msg):
+        self.assert_function_fail(service.START, msg)
+
+    def assert_action_stop_fail(self, msg):
+        self.assert_function_fail(service.STOP, msg)
+
+    def assert_function_fail(self, action, msg):
+        expected_error = "Action '{}' failed: {}".format(action, msg)
+        self.function_fail.assert_called_with(expected_error)
+
+    @staticmethod
+    def call_action_start():
+        service.main(['start'])
+
+    @staticmethod
+    def call_action_stop():
+        service.main(['stop'])
+
+    def test_systemctl_execute_all(self):
+        action = 'start'
+        services = service.ALL
+
+        expected_call = mock.call(['systemctl', action, self._TARGET_ALL],
+                                  timeout=self._CHECK_CALL_TIMEOUT)
+
+        service.systemctl_execute(action, services)
+
+        self.subprocess.check_call.assert_has_calls([expected_call])
+
+    def systemctl_execute_specific(self):
+        action = 'start'
+        services = ['ceph-osd@1.service', 'ceph-osd@2.service']
+
+        systemctl_call = ['systemctl', action] + services
+        expected_call = mock.call(systemctl_call,
+                                  timeout=self._CHECK_CALL_TIMEOUT)
+
+        service.systemctl_execute(action, services)
+
+        self.subprocess.check_call.assert_has_calls([expected_call])
+
+    def test_id_translation(self):
+        service_ids = {1, service.ALL, 2}
+        expected_names = [
+            'ceph-osd@1.service',
+            service.ALL,
+            'ceph-osd@2.service',
+        ]
+        service_names = service.osd_ids_to_service_names(service_ids)
+        self.assertEqual(sorted(service_names), sorted(expected_names))
+
+    def test_skip_service_presence_check(self):
+        service_list = [service.ALL]
+
+        service.check_service_is_present(service_list)
+
+        self.subprocess.run.assert_not_called()
+
+    def test_raise_all_missing_services(self):
+        missing_service_id = '99,100'
+        missing_list = []
+        for id_ in missing_service_id.split(','):
+            missing_list.append("ceph-osd@{}.service".format(id_))
+
+        service_list_cmd = ['systemctl', 'list-units', '--full', '--all',
+                            '--no-pager', '-t', 'service']
+
+        err_msg = 'Some services are not present on this ' \
+                  'unit: {}'.format(missing_list)
+
+        with self.assertRaises(RuntimeError, msg=err_msg):
+            service.check_service_is_present(missing_list)
+
+        self.subprocess.run.assert_called_with(service_list_cmd,
+                                               stdout=self.subprocess.PIPE,
+                                               timeout=30)
+
+    def test_raise_on_missing_arguments(self):
+        err_msg = 'Action argument "osds" is missing'
+        with self.func_call_arguments(osds=None):
+            with self.assertRaises(RuntimeError, msg=err_msg):
+                service.parse_arguments()
+
+    def test_parse_service_ids(self):
+        raw = '1,2,3'
+        expected_ids = {'1', '2', '3'}
+
+        with self.func_call_arguments(osds=raw):
+            parsed = service.parse_arguments()
+            self.assertEqual(parsed, expected_ids)
+
+    def test_parse_service_ids_with_all(self):
+        raw = '1,2,all'
+        expected_id = {service.ALL}
+
+        with self.func_call_arguments(osds=raw):
+            parsed = service.parse_arguments()
+            self.assertEqual(parsed, expected_id)
+
+    def test_fail_execute_unknown_action(self):
+        action = 'foo'
+        err_msg = 'Unknown action "{}"'.format(action)
+        with self.assertRaises(RuntimeError, msg=err_msg):
+            service.execute_action(action)
+
+    @mock.patch.object(service, 'systemctl_execute')
+    def test_execute_action(self, _):
+        with self.func_call_arguments(osds=service.ALL):
+            service.execute_action(service.START)
+            service.systemctl_execute.assert_called_with(service.START,
+                                                         [service.ALL])
+
+            service.execute_action(service.STOP)
+            service.systemctl_execute.assert_called_with(service.STOP,
+                                                         [service.ALL])
+
+    @mock.patch.object(service, 'execute_action')
+    def test_action_stop(self, execute_action):
+        self.call_action_stop()
+        execute_action.assert_called_with(service.STOP)
+
+    @mock.patch.object(service, 'execute_action')
+    def test_action_start(self, execute_action):
+        self.call_action_start()
+        execute_action.assert_called_with(service.START)
+
+    def test_actions_requires_systemd(self):
+        """Actions will fail if systemd is not present on the system"""
+        self.shutil.which.return_value = None
+        expected_error = 'This action requires systemd'
+        with self.func_call_arguments(osds='all'):
+            self.call_action_start()
+            self.assert_action_start_fail(expected_error)
+
+            self.call_action_stop()
+            self.assert_action_stop_fail(expected_error)
+
+            self.subprocess.check_call.assert_not_called()
+
+    def test_unknown_action(self):
+        action = 'foo'
+        err_msg = 'Action {} undefined'.format(action)
+        service.main([action])
+        self.function_fail.assert_called_with(err_msg)
+
+    @mock.patch.object(service, 'execute_action')
+    def test_action_failure(self, start_function):
+        err_msg = 'Test Error'
+        service.execute_action.side_effect = RuntimeError(err_msg)
+
+        self.call_action_start()
+
+        self.assert_action_start_fail(err_msg)