Merge lp:~niedbalski/charms/trusty/memcached/replication into lp:charms/trusty/memcached

[Jorge Niedbalski]

This change adds server side replication to memcached charm by using the repcached patches (http://repcached.lab.klab.org) for fixing bug https://bugs.launchpad.net/charms/+source/memcached/+bug/1421660

- Added the peer cluster relation
- Updated Readme
- Added unit/functional tests

Please note that replication only works betweeen 2 units.

[Billy Olsen (billy-olsen)]

A few minor comments inline. Overall looks pretty good. It'd be nice if the replication worked beyond 2 units, but it is what it is.

review: Needs Fixing

[Felipe Reyes (freyes)]

Jorge,

Your patch looks OK to me, the replication worked fine after the changes you made. Functional/unit tests are passing.

This is a good addition to the charm :)

Thanks,

review: Approve

=== modified file 'README.md'
--- README.md	2014-01-15 20:04:13 +0000
+++ README.md	2015-02-18 02:12:32 +0000
@@ -18,6 +18,45 @@
 
 The "3" in this case is an example, the `juju status memcached` will show you which machine number the service is running on so you can `juju ssh` to it.
 
+# Replication
+
+The charm uses the repcached patch ( http://repcached.lab.klab.org/ ), this patch has some limitations, as
+the ability to just replicate between 2 nodes.
+
+For enabling replication create a config.yaml file with the following content:
+    ```yaml
+    memcached:
+        repcached: True
+    ```
+
+Then deploy a maximum of 2 units of memcached:
+
+    juju deploy -n 2 config.yaml memcached
+
+**Caution** : As per design limitations, If you try to add another unit of memcached, all the units will be 
+set as standalone you decide to disable repcached by using `juju set memcached repcached=false` and re-deploy the
+unit.
+
+Or you can deploy 2 units and then enable replication by running
+    juju deploy -n 2 memcached
+    juju set memcached repcached=true
+
+
+### Removing a unit
+
+Removing one of the cluster units, means remove replication, please disable replication first
+and then remove the unit safely.
+
+    juju set memcached repcached=false
+    juju remove-unit memcached/0
+
+
+### Removing replication
+
+For turning the replication support off on memcached, you need to run the following command:
+
+    juju set memcached repcached=false
+
 ## Example Usage
 
 This charm can be used with other charms, in particular make note of [these possible relations](https://jujucharms.com/fullscreen/search/precise/memcached-7/?text=memcached#bws-related-charms)
@@ -26,11 +65,11 @@
 
 ## Scale out Usage
 
-You can 
+You can
 
     juju add-unit memcached
 
-To add more units. Memcached doesn't share load, it's very simple and the clients have the intelligence to know which server to pick. 
+To add more units. Memcached doesn't share load, it's very simple and the clients have the intelligence to know which server to pick.
 
 ## Known Limitations and Issues
 

=== modified file 'charm-helpers.yaml'
--- charm-helpers.yaml	2014-12-04 18:52:50 +0000
+++ charm-helpers.yaml	2015-02-18 02:12:32 +0000
@@ -4,3 +4,4 @@
     - fetch
     - core
     - contrib.network
+    - contrib.hahelpers

=== modified file 'config.yaml'
--- config.yaml	2014-12-01 19:57:56 +0000
+++ config.yaml	2015-02-18 02:12:32 +0000
@@ -75,16 +75,31 @@
         type: boolean
         default: False
     extra-options:
-        description: memcached has many other options documented in its man page. You may pass them here as a string which will be appended to memcached's execution.
-        type: string
-        default: ""
+      description: memcached has many other options documented in its man page. You may pass them here as a string which
+      type: string
+      default: ""
     nagios_context:
         default: "juju"
         type: string
-        description: >
+        description: |
             Used by the nrpe-external-master subordinate charm.
             A string that will be prepended to instance name to set the host name
             in nagios. So for instance the hostname would be something like:
                 juju-memcached-0
             If you're running multiple environments with the same services in them
             this allows you to differentiate between them.
+    repcached:
+      default: False
+      type: boolean
+      description: |
+        Enable memcached replication
+    repcached_origin:
+      default: "ppa:niedbalski/memcached-repcached"
+      type: string
+      description: |
+        Memcached + Repcached package location
+    repcached_port:
+      default: "11212"
+      type: string
+      description: |
+        TCP port number for replication (default: 11212)

=== added directory 'hooks/charmhelpers/contrib/hahelpers'
=== added file 'hooks/charmhelpers/contrib/hahelpers/__init__.py'
--- hooks/charmhelpers/contrib/hahelpers/__init__.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/contrib/hahelpers/__init__.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,15 @@
+# Copyright 2014-2015 Canonical Limited.
+#
+# This file is part of charm-helpers.
+#
+# charm-helpers is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3 as
+# published by the Free Software Foundation.
+#
+# charm-helpers is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.

=== added file 'hooks/charmhelpers/contrib/hahelpers/apache.py'
--- hooks/charmhelpers/contrib/hahelpers/apache.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/contrib/hahelpers/apache.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,82 @@
+# Copyright 2014-2015 Canonical Limited.
+#
+# This file is part of charm-helpers.
+#
+# charm-helpers is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3 as
+# published by the Free Software Foundation.
+#
+# charm-helpers is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
+
+#
+# Copyright 2012 Canonical Ltd.
+#
+# This file is sourced from lp:openstack-charm-helpers
+#
+# Authors:
+#  James Page <[e-mail]>
+#  Adam Gandelman <[e-mail]>
+#
+
+import subprocess
+
+from charmhelpers.core.hookenv import (
+    config as config_get,
+    relation_get,
+    relation_ids,
+    related_units as relation_list,
+    log,
+    INFO,
+)
+
+
+def get_cert(cn=None):
+    # TODO: deal with multiple https endpoints via charm config
+    cert = config_get('ssl_cert')
+    key = config_get('ssl_key')
+    if not (cert and key):
+        log("Inspecting identity-service relations for SSL certificate.",
+            level=INFO)
+        cert = key = None
+        if cn:
+            ssl_cert_attr = 'ssl_cert_{}'.format(cn)
+            ssl_key_attr = 'ssl_key_{}'.format(cn)
+        else:
+            ssl_cert_attr = 'ssl_cert'
+            ssl_key_attr = 'ssl_key'
+        for r_id in relation_ids('identity-service'):
+            for unit in relation_list(r_id):
+                if not cert:
+                    cert = relation_get(ssl_cert_attr,
+                                        rid=r_id, unit=unit)
+                if not key:
+                    key = relation_get(ssl_key_attr,
+                                       rid=r_id, unit=unit)
+    return (cert, key)
+
+
+def get_ca_cert():
+    ca_cert = config_get('ssl_ca')
+    if ca_cert is None:
+        log("Inspecting identity-service relations for CA SSL certificate.",
+            level=INFO)
+        for r_id in relation_ids('identity-service'):
+            for unit in relation_list(r_id):
+                if ca_cert is None:
+                    ca_cert = relation_get('ca_cert',
+                                           rid=r_id, unit=unit)
+    return ca_cert
+
+
+def install_ca_cert(ca_cert):
+    if ca_cert:
+        with open('/usr/local/share/ca-certificates/keystone_juju_ca_cert.crt',
+                  'w') as crt:
+            crt.write(ca_cert)
+        subprocess.check_call(['update-ca-certificates', '--fresh'])

=== added file 'hooks/charmhelpers/contrib/hahelpers/cluster.py'
--- hooks/charmhelpers/contrib/hahelpers/cluster.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/contrib/hahelpers/cluster.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,268 @@
+# Copyright 2014-2015 Canonical Limited.
+#
+# This file is part of charm-helpers.
+#
+# charm-helpers is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3 as
+# published by the Free Software Foundation.
+#
+# charm-helpers is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
+
+#
+# Copyright 2012 Canonical Ltd.
+#
+# Authors:
+#  James Page <[e-mail]>
+#  Adam Gandelman <[e-mail]>
+#
+
+"""
+Helpers for clustering and determining "cluster leadership" and other
+clustering-related helpers.
+"""
+
+import subprocess
+import os
+
+from socket import gethostname as get_unit_hostname
+
+import six
+
+from charmhelpers.core.hookenv import (
+    log,
+    relation_ids,
+    related_units as relation_list,
+    relation_get,
+    config as config_get,
+    INFO,
+    ERROR,
+    WARNING,
+    unit_get,
+)
+from charmhelpers.core.decorators import (
+    retry_on_exception,
+)
+
+
+class HAIncompleteConfig(Exception):
+    pass
+
+
+class CRMResourceNotFound(Exception):
+    pass
+
+
+def is_elected_leader(resource):
+    """
+    Returns True if the charm executing this is the elected cluster leader.
+
+    It relies on two mechanisms to determine leadership:
+        1. If the charm is part of a corosync cluster, call corosync to
+        determine leadership.
+        2. If the charm is not part of a corosync cluster, the leader is
+        determined as being "the alive unit with the lowest unit numer". In
+        other words, the oldest surviving unit.
+    """
+    if is_clustered():
+        if not is_crm_leader(resource):
+            log('Deferring action to CRM leader.', level=INFO)
+            return False
+    else:
+        peers = peer_units()
+        if peers and not oldest_peer(peers):
+            log('Deferring action to oldest service unit.', level=INFO)
+            return False
+    return True
+
+
+def is_clustered():
+    for r_id in (relation_ids('ha') or []):
+        for unit in (relation_list(r_id) or []):
+            clustered = relation_get('clustered',
+                                     rid=r_id,
+                                     unit=unit)
+            if clustered:
+                return True
+    return False
+
+
+@retry_on_exception(5, base_delay=2, exc_type=CRMResourceNotFound)
+def is_crm_leader(resource, retry=False):
+    """
+    Returns True if the charm calling this is the elected corosync leader,
+    as returned by calling the external "crm" command.
+
+    We allow this operation to be retried to avoid the possibility of getting a
+    false negative. See LP #1396246 for more info.
+    """
+    cmd = ['crm', 'resource', 'show', resource]
+    try:
+        status = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
+        if not isinstance(status, six.text_type):
+            status = six.text_type(status, "utf-8")
+    except subprocess.CalledProcessError:
+        status = None
+
+    if status and get_unit_hostname() in status:
+        return True
+
+    if status and "resource %s is NOT running" % (resource) in status:
+        raise CRMResourceNotFound("CRM resource %s not found" % (resource))
+
+    return False
+
+
+def is_leader(resource):
+    log("is_leader is deprecated. Please consider using is_crm_leader "
+        "instead.", level=WARNING)
+    return is_crm_leader(resource)
+
+
+def peer_units(peer_relation="cluster"):
+    peers = []
+    for r_id in (relation_ids(peer_relation) or []):
+        for unit in (relation_list(r_id) or []):
+            peers.append(unit)
+    return peers
+
+
+def peer_ips(peer_relation='cluster', addr_key='private-address'):
+    '''Return a dict of peers and their private-address'''
+    peers = {}
+    for r_id in relation_ids(peer_relation):
+        for unit in relation_list(r_id):
+            peers[unit] = relation_get(addr_key, rid=r_id, unit=unit)
+    return peers
+
+
+def oldest_peer(peers):
+    """Determines who the oldest peer is by comparing unit numbers."""
+    local_unit_no = int(os.getenv('JUJU_UNIT_NAME').split('/')[1])
+    for peer in peers:
+        remote_unit_no = int(peer.split('/')[1])
+        if remote_unit_no < local_unit_no:
+            return False
+    return True
+
+
+def eligible_leader(resource):
+    log("eligible_leader is deprecated. Please consider using "
+        "is_elected_leader instead.", level=WARNING)
+    return is_elected_leader(resource)
+
+
+def https():
+    '''
+    Determines whether enough data has been provided in configuration
+    or relation data to configure HTTPS
+    .
+    returns: boolean
+    '''
+    if config_get('use-https') == "yes":
+        return True
+    if config_get('ssl_cert') and config_get('ssl_key'):
+        return True
+    for r_id in relation_ids('identity-service'):
+        for unit in relation_list(r_id):
+            # TODO - needs fixing for new helper as ssl_cert/key suffixes with CN
+            rel_state = [
+                relation_get('https_keystone', rid=r_id, unit=unit),
+                relation_get('ca_cert', rid=r_id, unit=unit),
+            ]
+            # NOTE: works around (LP: #1203241)
+            if (None not in rel_state) and ('' not in rel_state):
+                return True
+    return False
+
+
+def determine_api_port(public_port, singlenode_mode=False):
+    '''
+    Determine correct API server listening port based on
+    existence of HTTPS reverse proxy and/or haproxy.
+
+    public_port: int: standard public port for given service
+
+    singlenode_mode: boolean: Shuffle ports when only a single unit is present
+
+    returns: int: the correct listening port for the API service
+    '''
+    i = 0
+    if singlenode_mode:
+        i += 1
+    elif len(peer_units()) > 0 or is_clustered():
+        i += 1
+    if https():
+        i += 1
+    return public_port - (i * 10)
+
+
+def determine_apache_port(public_port, singlenode_mode=False):
+    '''
+    Description: Determine correct apache listening port based on public IP +
+    state of the cluster.
+
+    public_port: int: standard public port for given service
+
+    singlenode_mode: boolean: Shuffle ports when only a single unit is present
+
+    returns: int: the correct listening port for the HAProxy service
+    '''
+    i = 0
+    if singlenode_mode:
+        i += 1
+    elif len(peer_units()) > 0 or is_clustered():
+        i += 1
+    return public_port - (i * 10)
+
+
+def get_hacluster_config(exclude_keys=None):
+    '''
+    Obtains all relevant configuration from charm configuration required
+    for initiating a relation to hacluster:
+
+        ha-bindiface, ha-mcastport, vip
+
+    param: exclude_keys: list of setting key(s) to be excluded.
+    returns: dict: A dict containing settings keyed by setting name.
+    raises: HAIncompleteConfig if settings are missing.
+    '''
+    settings = ['ha-bindiface', 'ha-mcastport', 'vip']
+    conf = {}
+    for setting in settings:
+        if exclude_keys and setting in exclude_keys:
+            continue
+
+        conf[setting] = config_get(setting)
+    missing = []
+    [missing.append(s) for s, v in six.iteritems(conf) if v is None]
+    if missing:
+        log('Insufficient config data to configure hacluster.', level=ERROR)
+        raise HAIncompleteConfig
+    return conf
+
+
+def canonical_url(configs, vip_setting='vip'):
+    '''
+    Returns the correct HTTP URL to this host given the state of HTTPS
+    configuration and hacluster.
+
+    :configs    : OSTemplateRenderer: A config tempating object to inspect for
+                                      a complete https context.
+
+    :vip_setting:                str: Setting in charm config that specifies
+                                      VIP address.
+    '''
+    scheme = 'http'
+    if 'https' in configs.complete_contexts():
+        scheme = 'https'
+    if is_clustered():
+        addr = config_get(vip_setting)
+    else:
+        addr = unit_get('private-address')
+    return '%s://%s' % (scheme, addr)

=== added file 'hooks/charmhelpers/core/unitdata.py'
--- hooks/charmhelpers/core/unitdata.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/core/unitdata.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,477 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2014-2015 Canonical Limited.
+#
+# This file is part of charm-helpers.
+#
+# charm-helpers is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3 as
+# published by the Free Software Foundation.
+#
+# charm-helpers is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
+#
+#
+# Authors:
+#  Kapil Thangavelu <[e-mail]>
+#
+"""
+Intro
+-----
+
+A simple way to store state in units. This provides a key value
+storage with support for versioned, transactional operation,
+and can calculate deltas from previous values to simplify unit logic
+when processing changes.
+
+
+Hook Integration
+----------------
+
+There are several extant frameworks for hook execution, including
+
+ - charmhelpers.core.hookenv.Hooks
+ - charmhelpers.core.services.ServiceManager
+
+The storage classes are framework agnostic, one simple integration is
+via the HookData contextmanager. It will record the current hook
+execution environment (including relation data, config data, etc.),
+setup a transaction and allow easy access to the changes from
+previously seen values. One consequence of the integration is the
+reservation of particular keys ('rels', 'unit', 'env', 'config',
+'charm_revisions') for their respective values.
+
+Here's a fully worked integration example using hookenv.Hooks::
+
+       from charmhelper.core import hookenv, unitdata
+
+       hook_data = unitdata.HookData()
+       db = unitdata.kv()
+       hooks = hookenv.Hooks()
+
+       @hooks.hook
+       def config_changed():
+           # Print all changes to configuration from previously seen
+           # values.
+           for changed, (prev, cur) in hook_data.conf.items():
+               print('config changed', changed,
+                     'previous value', prev,
+                     'current value',  cur)
+
+           # Get some unit specific bookeeping
+           if not db.get('pkg_key'):
+               key = urllib.urlopen('https://example.com/pkg_key').read()
+               db.set('pkg_key', key)
+
+           # Directly access all charm config as a mapping.
+           conf = db.getrange('config', True)
+
+           # Directly access all relation data as a mapping
+           rels = db.getrange('rels', True)
+
+       if __name__ == '__main__':
+           with hook_data():
+               hook.execute()
+
+
+A more basic integration is via the hook_scope context manager which simply
+manages transaction scope (and records hook name, and timestamp)::
+
+  >>> from unitdata import kv
+  >>> db = kv()
+  >>> with db.hook_scope('install'):
+  ...    # do work, in transactional scope.
+  ...    db.set('x', 1)
+  >>> db.get('x')
+  1
+
+
+Usage
+-----
+
+Values are automatically json de/serialized to preserve basic typing
+and complex data struct capabilities (dicts, lists, ints, booleans, etc).
+
+Individual values can be manipulated via get/set::
+
+   >>> kv.set('y', True)
+   >>> kv.get('y')
+   True
+
+   # We can set complex values (dicts, lists) as a single key.
+   >>> kv.set('config', {'a': 1, 'b': True'})
+
+   # Also supports returning dictionaries as a record which
+   # provides attribute access.
+   >>> config = kv.get('config', record=True)
+   >>> config.b
+   True
+
+
+Groups of keys can be manipulated with update/getrange::
+
+   >>> kv.update({'z': 1, 'y': 2}, prefix="gui.")
+   >>> kv.getrange('gui.', strip=True)
+   {'z': 1, 'y': 2}
+
+When updating values, its very helpful to understand which values
+have actually changed and how have they changed. The storage
+provides a delta method to provide for this::
+
+   >>> data = {'debug': True, 'option': 2}
+   >>> delta = kv.delta(data, 'config.')
+   >>> delta.debug.previous
+   None
+   >>> delta.debug.current
+   True
+   >>> delta
+   {'debug': (None, True), 'option': (None, 2)}
+
+Note the delta method does not persist the actual change, it needs to
+be explicitly saved via 'update' method::
+
+   >>> kv.update(data, 'config.')
+
+Values modified in the context of a hook scope retain historical values
+associated to the hookname.
+
+   >>> with db.hook_scope('config-changed'):
+   ...      db.set('x', 42)
+   >>> db.gethistory('x')
+   [(1, u'x', 1, u'install', u'2015-01-21T16:49:30.038372'),
+    (2, u'x', 42, u'config-changed', u'2015-01-21T16:49:30.038786')]
+
+"""
+
+import collections
+import contextlib
+import datetime
+import json
+import os
+import pprint
+import sqlite3
+import sys
+
+__author__ = 'Kapil Thangavelu <[e-mail]>'
+
+
+class Storage(object):
+    """Simple key value database for local unit state within charms.
+
+    Modifications are automatically committed at hook exit. That's
+    currently regardless of exit code.
+
+    To support dicts, lists, integer, floats, and booleans values
+    are automatically json encoded/decoded.
+    """
+    def __init__(self, path=None):
+        self.db_path = path
+        if path is None:
+            self.db_path = os.path.join(
+                os.environ.get('CHARM_DIR', ''), '.unit-state.db')
+        self.conn = sqlite3.connect('%s' % self.db_path)
+        self.cursor = self.conn.cursor()
+        self.revision = None
+        self._closed = False
+        self._init()
+
+    def close(self):
+        if self._closed:
+            return
+        self.flush(False)
+        self.cursor.close()
+        self.conn.close()
+        self._closed = True
+
+    def _scoped_query(self, stmt, params=None):
+        if params is None:
+            params = []
+        return stmt, params
+
+    def get(self, key, default=None, record=False):
+        self.cursor.execute(
+            *self._scoped_query(
+                'select data from kv where key=?', [key]))
+        result = self.cursor.fetchone()
+        if not result:
+            return default
+        if record:
+            return Record(json.loads(result[0]))
+        return json.loads(result[0])
+
+    def getrange(self, key_prefix, strip=False):
+        stmt = "select key, data from kv where key like '%s%%'" % key_prefix
+        self.cursor.execute(*self._scoped_query(stmt))
+        result = self.cursor.fetchall()
+
+        if not result:
+            return None
+        if not strip:
+            key_prefix = ''
+        return dict([
+            (k[len(key_prefix):], json.loads(v)) for k, v in result])
+
+    def update(self, mapping, prefix=""):
+        for k, v in mapping.items():
+            self.set("%s%s" % (prefix, k), v)
+
+    def unset(self, key):
+        self.cursor.execute('delete from kv where key=?', [key])
+        if self.revision and self.cursor.rowcount:
+            self.cursor.execute(
+                'insert into kv_revisions values (?, ?, ?)',
+                [key, self.revision, json.dumps('DELETED')])
+
+    def set(self, key, value):
+        serialized = json.dumps(value)
+
+        self.cursor.execute(
+            'select data from kv where key=?', [key])
+        exists = self.cursor.fetchone()
+
+        # Skip mutations to the same value
+        if exists:
+            if exists[0] == serialized:
+                return value
+
+        if not exists:
+            self.cursor.execute(
+                'insert into kv (key, data) values (?, ?)',
+                (key, serialized))
+        else:
+            self.cursor.execute('''
+            update kv
+            set data = ?
+            where key = ?''', [serialized, key])
+
+        # Save
+        if not self.revision:
+            return value
+
+        self.cursor.execute(
+            'select 1 from kv_revisions where key=? and revision=?',
+            [key, self.revision])
+        exists = self.cursor.fetchone()
+
+        if not exists:
+            self.cursor.execute(
+                '''insert into kv_revisions (
+                revision, key, data) values (?, ?, ?)''',
+                (self.revision, key, serialized))
+        else:
+            self.cursor.execute(
+                '''
+                update kv_revisions
+                set data = ?
+                where key = ?
+                and   revision = ?''',
+                [serialized, key, self.revision])
+
+        return value
+
+    def delta(self, mapping, prefix):
+        """
+        return a delta containing values that have changed.
+        """
+        previous = self.getrange(prefix, strip=True)
+        if not previous:
+            pk = set()
+        else:
+            pk = set(previous.keys())
+        ck = set(mapping.keys())
+        delta = DeltaSet()
+
+        # added
+        for k in ck.difference(pk):
+            delta[k] = Delta(None, mapping[k])
+
+        # removed
+        for k in pk.difference(ck):
+            delta[k] = Delta(previous[k], None)
+
+        # changed
+        for k in pk.intersection(ck):
+            c = mapping[k]
+            p = previous[k]
+            if c != p:
+                delta[k] = Delta(p, c)
+
+        return delta
+
+    @contextlib.contextmanager
+    def hook_scope(self, name=""):
+        """Scope all future interactions to the current hook execution
+        revision."""
+        assert not self.revision
+        self.cursor.execute(
+            'insert into hooks (hook, date) values (?, ?)',
+            (name or sys.argv[0],
+             datetime.datetime.utcnow().isoformat()))
+        self.revision = self.cursor.lastrowid
+        try:
+            yield self.revision
+            self.revision = None
+        except:
+            self.flush(False)
+            self.revision = None
+            raise
+        else:
+            self.flush()
+
+    def flush(self, save=True):
+        if save:
+            self.conn.commit()
+        elif self._closed:
+            return
+        else:
+            self.conn.rollback()
+
+    def _init(self):
+        self.cursor.execute('''
+            create table if not exists kv (
+               key text,
+               data text,
+               primary key (key)
+               )''')
+        self.cursor.execute('''
+            create table if not exists kv_revisions (
+               key text,
+               revision integer,
+               data text,
+               primary key (key, revision)
+               )''')
+        self.cursor.execute('''
+            create table if not exists hooks (
+               version integer primary key autoincrement,
+               hook text,
+               date text
+               )''')
+        self.conn.commit()
+
+    def gethistory(self, key, deserialize=False):
+        self.cursor.execute(
+            '''
+            select kv.revision, kv.key, kv.data, h.hook, h.date
+            from kv_revisions kv,
+                 hooks h
+            where kv.key=?
+             and kv.revision = h.version
+            ''', [key])
+        if deserialize is False:
+            return self.cursor.fetchall()
+        return map(_parse_history, self.cursor.fetchall())
+
+    def debug(self, fh=sys.stderr):
+        self.cursor.execute('select * from kv')
+        pprint.pprint(self.cursor.fetchall(), stream=fh)
+        self.cursor.execute('select * from kv_revisions')
+        pprint.pprint(self.cursor.fetchall(), stream=fh)
+
+
+def _parse_history(d):
+    return (d[0], d[1], json.loads(d[2]), d[3],
+            datetime.datetime.strptime(d[-1], "%Y-%m-%dT%H:%M:%S.%f"))
+
+
+class HookData(object):
+    """Simple integration for existing hook exec frameworks.
+
+    Records all unit information, and stores deltas for processing
+    by the hook.
+
+    Sample::
+
+       from charmhelper.core import hookenv, unitdata
+
+       changes = unitdata.HookData()
+       db = unitdata.kv()
+       hooks = hookenv.Hooks()
+
+       @hooks.hook
+       def config_changed():
+           # View all changes to configuration
+           for changed, (prev, cur) in changes.conf.items():
+               print('config changed', changed,
+                     'previous value', prev,
+                     'current value',  cur)
+
+           # Get some unit specific bookeeping
+           if not db.get('pkg_key'):
+               key = urllib.urlopen('https://example.com/pkg_key').read()
+               db.set('pkg_key', key)
+
+       if __name__ == '__main__':
+           with changes():
+               hook.execute()
+
+    """
+    def __init__(self):
+        self.kv = kv()
+        self.conf = None
+        self.rels = None
+
+    @contextlib.contextmanager
+    def __call__(self):
+        from charmhelpers.core import hookenv
+        hook_name = hookenv.hook_name()
+
+        with self.kv.hook_scope(hook_name):
+            self._record_charm_version(hookenv.charm_dir())
+            delta_config, delta_relation = self._record_hook(hookenv)
+            yield self.kv, delta_config, delta_relation
+
+    def _record_charm_version(self, charm_dir):
+        # Record revisions.. charm revisions are meaningless
+        # to charm authors as they don't control the revision.
+        # so logic dependnent on revision is not particularly
+        # useful, however it is useful for debugging analysis.
+        charm_rev = open(
+            os.path.join(charm_dir, 'revision')).read().strip()
+        charm_rev = charm_rev or '0'
+        revs = self.kv.get('charm_revisions', [])
+        if charm_rev not in revs:
+            revs.append(charm_rev.strip() or '0')
+            self.kv.set('charm_revisions', revs)
+
+    def _record_hook(self, hookenv):
+        data = hookenv.execution_environment()
+        self.conf = conf_delta = self.kv.delta(data['conf'], 'config')
+        self.rels = rels_delta = self.kv.delta(data['rels'], 'rels')
+        self.kv.set('env', data['env'])
+        self.kv.set('unit', data['unit'])
+        self.kv.set('relid', data.get('relid'))
+        return conf_delta, rels_delta
+
+
+class Record(dict):
+
+    __slots__ = ()
+
+    def __getattr__(self, k):
+        if k in self:
+            return self[k]
+        raise AttributeError(k)
+
+
+class DeltaSet(Record):
+
+    __slots__ = ()
+
+
+Delta = collections.namedtuple('Delta', ['previous', 'current'])
+
+
+_KV = None
+
+
+def kv():
+    global _KV
+    if _KV is None:
+        _KV = Storage()
+    return _KV

=== added symlink 'hooks/cluster-relation-changed'
=== target is u'memcached_hooks.py'
=== added symlink 'hooks/cluster-relation-joined'
=== target is u'memcached_hooks.py'
=== modified file 'hooks/memcached_hooks.py'
--- hooks/memcached_hooks.py	2014-12-19 18:24:29 +0000
+++ hooks/memcached_hooks.py	2015-02-18 02:12:32 +0000
@@ -1,11 +1,10 @@
 #!/usr/bin/env python
-__author__ = 'Felipe Reyes <[e-mail]>'
-
 import re
 import os
 import shutil
 import subprocess
 import sys
+import time
 
 from charmhelpers.core.hookenv import (
     config,
@@ -25,10 +24,19 @@
     service_start,
     service_stop,
 )
-from charmhelpers.fetch import apt_install, apt_update
+
+from charmhelpers.contrib.hahelpers.cluster import (
+    oldest_peer,
+    peer_units,
+)
+
+from charmhelpers.fetch import apt_install, apt_update, add_source
 from charmhelpers.contrib.network import ufw
+
 import memcached_utils
+import replication
 
+__author__ = 'Felipe Reyes <[e-mail]>'
 
 DOT = os.path.dirname(os.path.abspath(__file__))
 ETC_DEFAULT_MEMCACHED = '/etc/default/memcached'
@@ -38,6 +46,7 @@
 MEMCACHE_NAGIOS_PLUGIN = '/usr/local/lib/nagios/plugins/check_memcache.py'
 RESTART_MAP = {ETC_MEMCACHED_CONF: ['memcached'],
                ETC_MUNIN_NODE_CONF: ['munin-node']}
+
 hooks = Hooks()
 
 
@@ -47,6 +56,11 @@
 
 @hooks.hook('install')
 def install():
+    enable_repcached = config('repcached')
+
+    if enable_repcached:
+        add_source(config('repcached_origin'))
+
     apt_update(fatal=True)
     apt_install(["memcached", "python-cheetah", "python-memcache"], fatal=True)
 
@@ -54,6 +68,10 @@
         f.write('ENABLE_MEMCACHED=yes\n')
 
     ufw.enable()
+
+    if enable_repcached:
+        ufw.service(config('repcached_port'), 'open')
+
     ufw.service('ssh', 'open')
 
 
@@ -67,6 +85,61 @@
     service_stop('memcached')
 
 
[e-mail]('cluster-relation-joined')
+def cluster_relation_joined():
+
+    if not config('repcached'):
+        return config_changed()
+
+    peers = peer_units()
+    if peers and not oldest_peer(peers):
+        log("Delaying operation to oldest peer", level='INFO')
+        return
+
+    replica = replication.get_repcached_replica()
+    if not replica:
+        replica = replication.set_repcached_replica()
+
+    master, replica = replica
+    log("Setting replication with peer unit: %s" % replica, level="INFO")
+    config_changed(replica=replica)
+
+
[e-mail]('cluster-relation-changed')
+def cluster_relation_changed():
+
+    if not config('repcached'):
+        return config_changed()
+
+    peers = peer_units()
+    if peers and oldest_peer(peers):
+        log("Delaying operation to secondary peer", level='INFO')
+        return
+
+    retries = 0
+    while retries < 3:
+        log('Waiting for master memcached node to become available',
+            level='INFO')
+        secondary = relation_get('replica')
+        try:
+            if secondary == unit_get('private-address'):
+                replication.store_replica(relation_get('master'), secondary)
+                return config_changed(replica=relation_get('master'))
+            elif secondary is None:
+                raise AttributeError()
+            else:
+                msg = """This unit will be marked as failed because
+                Repcached replication only can take place between two units.
+                If you want to disable this set 'juju set repcached=false'"""
+                log(msg, level='WARN')
+                raise Exception(msg)
+        except AttributeError:
+            log("Master not yet available, waiting for %d secs" %
+                replication.REPCACHED_MASTER_WAIT)
+            time.sleep(replication.REPCACHED_MASTER_WAIT)
+            retries += 1
+
+
 @hooks.hook('cache-relation-joined')
 def cache_relation_joined():
 
@@ -92,7 +165,7 @@
 
 @hooks.hook('config-changed')
 @restart_on_change(RESTART_MAP)
-def config_changed():
+def config_changed(replica=None):
     mem_size = config('size')
 
     if mem_size == 0:
@@ -100,9 +173,38 @@
         mem_size = int(re.findall(r'\d+', output.split('\n')[2])[1])
         mem_size = int(mem_size * 0.9)
 
+    if config('repcached'):
+        # If repcached was enabled after install, we need
+        # to make sure to install the memcached package with replication
+        # enabled.
+        if not memcached_utils.dpkg_info_contains('memcached',
+                                                  'version',
+                                                  'repcache'):
+            add_source(config('repcached_origin'))
+            apt_update(fatal=True)
+            apt_install(["memcached"], fatal=True)
+
+        if not replica:
+            replica = replication.get_current_replica()
+
+    elif (
+        config('repcached') and
+        os.path.exists(replication.REPCACHED_REPLICA_FILE)
+    ):
+        log("Replication has been disabled, removing from this unit",
+            level='INFO')
+        ufw.service(config('repcached_port'), 'close')
+        os.remove(replication.REPCACHED_REPLICA_FILE)
+        replica = None
+
     configs = {'mem_size': mem_size,
+               'replica': replica,
                'large_pages_enabled': False}
 
+    if replica:
+        configs['repcached_port'] = config('repcached_port')
+        ufw.service(config('repcached_port'), 'open')
+
     for key in ['request-limit', 'min-item-size', 'slab-page-size', 'threads',
                 'disable-auto-cleanup', 'disable-cas', 'factor',
                 'connection-limit', 'tcp-port', 'udp-port']:

=== modified file 'hooks/memcached_utils.py'
--- hooks/memcached_utils.py	2014-12-04 19:57:52 +0000
+++ hooks/memcached_utils.py	2015-02-18 02:12:32 +0000
@@ -1,11 +1,13 @@
-__author__ = 'Felipe Reyes <[e-mail]>'
-
 from charmhelpers.contrib.network import ufw
 from charmhelpers.core.hookenv import (
     config,
     log,
 )
 
+import subprocess
+
+__author__ = 'Felipe Reyes <[e-mail]>'
+
 
 def munin_format_ip(ip):
     return "^{}$".format(ip.replace('.', '\\.'))
@@ -25,3 +27,16 @@
 
     if config('udp-port') > 0:
         ufw.revoke_access(address, port=str(config('udp-port')), proto='udp')
+
+
+def dpkg_info_contains(package, key, value):
+    info = {}
+    dpkg = subprocess.check_output(
+        ['dpkg', '-s', package]).splitlines()
+
+    for entry in dpkg:
+        entry = entry.split(':')
+        if len(entry) > 1:
+            info.update({entry[0].lower(): entry[1]})
+
+    return info.get(key, -1).find(value) != -1

=== added file 'hooks/replication.py'
--- hooks/replication.py	1970-01-01 00:00:00 +0000
+++ hooks/replication.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,84 @@
+from charmhelpers.core.hookenv import (
+    log,
+    relation_id,
+    relation_set,
+    unit_get,
+)
+
+from charmhelpers.contrib.hahelpers.cluster import (
+    oldest_peer,
+    peer_units,
+    peer_ips,
+)
+
+REPCACHED_REPLICA_FILE = "/var/run/repcached_replica"
+REPCACHED_MASTER_WAIT = 10
+
+
+def store_replica(master, replica):
+    with open(REPCACHED_REPLICA_FILE, 'w+') as fd:
+        fd.write("%s,%s" % (master, replica))
+
+
+def get_current_replica():
+    replica = get_repcached_replica()
+
+    if replica:
+        master, secondary = replica
+        ip_addr = unit_get('private-address')
+        if ip_addr == master:
+            replica = secondary
+        elif ip_addr == secondary:
+            replica = master
+    else:
+        peers = peer_units()
+
+        if len(peers) > 1:
+            msg = """Cannot configure replication between more than 2 units,
+            all the units are going to be setup as standalone. Please
+            remove the remaining units with 'juju remove-unit', and
+            then set 'juju set memcached repcached=true' again"""
+            log(msg, level='WARN')
+        else:
+            try:
+                peer_unit = peer_ips().values()[0]
+
+                if oldest_peer(peers):
+                    master, secondary = (unit_get('private-address'),
+                                         peer_unit)
+                    replica = secondary
+                else:
+                    master, secondary = (peer_unit,
+                                         unit_get('private-address'))
+                    replica = master
+
+                store_replica(master, secondary)
+
+            except IndexError:
+                replica = None
+
+    log("Current replica value: %s" % replica, level="INFO")
+    return replica
+
+
+def get_repcached_replica():
+    try:
+        with open(REPCACHED_REPLICA_FILE, 'r') as fd:
+            return tuple(fd.read().split(','))
+    except:
+        return None
+
+
+def set_repcached_replica():
+    replica = peer_ips().values()[0]
+    log("Setting replica unit: %s" % replica)
+
+    master = unit_get('private-address')
+    relation_set(relation_id(), {
+        'master': master,
+        'replica': replica,
+    })
+
+    with open(REPCACHED_REPLICA_FILE, 'w+') as fd:
+        fd.write("%s,%s" % (master, replica))
+    return (master, replica)

=== modified file 'metadata.yaml'
--- metadata.yaml	2014-12-09 17:19:45 +0000
+++ metadata.yaml	2015-02-18 02:12:32 +0000
@@ -1,7 +1,7 @@
 name: memcached
 summary: "A high-performance memory object caching system"
 maintainer: Felipe Reyes <[e-mail]>
-description: 
+description:
  memcached optimizes specific high-load serving applications that are designed
  to take advantage of its versatile no-locking memory access system. Clients
  are available in several different programming languages, to suit the needs
@@ -17,3 +17,6 @@
   nrpe-external-master:
     interface: nrpe-external-master
     scope: container
+peers:
+  cluster:
+    interface: memcached-replication
\ No newline at end of file

=== modified file 'revision'
--- revision	2013-04-18 16:23:01 +0000
+++ revision	2015-02-18 02:12:32 +0000
@@ -1,1 +1,1 @@
-28
+69

=== modified file 'templates/memcached.conf'
--- templates/memcached.conf	2014-12-01 20:53:36 +0000
+++ templates/memcached.conf	2015-02-18 02:12:32 +0000
@@ -8,7 +8,7 @@
 # memcached default config file
 # 2003 - Jay Bonci <[e-mail]>
 # This configuration file is read by the start-memcached script provided as
-# part of the Debian GNU/Linux distribution. 
+# part of the Debian GNU/Linux distribution.
 
 # Run memcached as a daemon. This command is implied, and is not needed for the
 # daemon to run. See the README.Debian that comes with this package for more
@@ -60,3 +60,5 @@
 {% if disable_auto_cleanup %}-M{% endif %}
 {% if disable_cas %}-C{% endif %}
 {% if factor != -1.0 %}-f {{factor}}{% endif %}
+{% if replica %}-x {{replica}}{% endif %}
+{% if repcached_port %}-X {{repcached_port}}{%endif%}
\ No newline at end of file

=== modified file 'tests/10_deploy_test.py'
--- tests/10_deploy_test.py	2014-12-09 17:19:45 +0000
+++ tests/10_deploy_test.py	2015-02-18 02:12:32 +0000
@@ -19,7 +19,7 @@
 
 # Create a configuration dictionary for custom memcached values.
 configuration = {'size': 512, 'connection-limit': 128, 'factor': 1.10,
-                 'tcp-port': 11212, 'udp-port': 11213}
+                 'tcp-port': 11214, 'udp-port': 11213}
 d.configure('memcached', configuration)
 # Expose memcached so it is visible to the tests.
 d.expose('memcached')

=== added file 'tests/20_deploy_replication_test.py'
--- tests/20_deploy_replication_test.py	1970-01-01 00:00:00 +0000
+++ tests/20_deploy_replication_test.py	2015-02-18 02:12:32 +0000
@@ -0,0 +1,117 @@
+#!/usr/bin/python3
+
+import amulet
+import telnetlib
+import time
+
+
+def check_config(a, b):
+    config_string = a.file_contents('/etc/memcached.conf')
+    # Parse the configuration file for the values sent in the deployment.
+    for line in config_string.splitlines():
+        print(line)
+        if line.startswith('-x'):
+            o = line.split()[1]
+            break
+
+    if o != b.info['public-address']:
+        amulet.raise_status(amulet.FAIL, msg='Incorrect replica address')
+
+
+def check_replication(a, b, port):
+    date_time = time.strftime("%F %r")
+    string = 'memcached test %s' % date_time
+
+    try:
+        tn = telnetlib.Telnet(a.info['public-address'], port)
+        command = 'set greeting 1 0 %d' % len(string)
+        tn.write(command.encode() + b'\r\n')
+        tn.write(string.encode() + b'\r\n')
+        response = tn.read_until(b'STORED', 2)
+        tn.close()
+
+        tn = telnetlib.Telnet(b.info['public-address'], port)
+        tn.write(b'get greeting\r\n')
+        response = tn.read_until(b'END', 2)
+        response = response.decode()
+        index = response.find(string)
+
+        if index != -1:
+            print('Found %s in the greeting response.' % string)
+        else:
+            print(response)
+            message = 'Did not find %s in the greeting from memcached' % string
+            amulet.raise_status(amulet.FAIL, msg=message)
+            tn.write(b'quit\n')
+
+    except Exception:
+        message = 'An error occurred communicating with memcached over telnet'
+        amulet.raise_status(amulet.FAIL, msg=message)
+    finally:
+        tn.close()
+
+
+seconds = 1200
+d = amulet.Deployment(series="trusty")
+d.add('memcached', units=2)
+
+configuration = {
+    'repcached': True,
+    'size': 512,
+    'connection-limit': 128,
+    'factor': 1.10,
+    'tcp-port': 11211,
+    'udp-port': 11213,
+}
+
+d.configure('memcached', configuration)
+d.expose('memcached')
+
+try:
+    d.setup(timeout=seconds)
+    d.sentry.wait(seconds)
+except amulet.helpers.TimeoutError:
+    message = 'The environment did not setup in %d seconds.' % seconds
+    amulet.raise_status(amulet.SKIP, msg=message)
+except:
+    raise
+
+units = (d.sentry['memcached/0'], d.sentry['memcached/1'])
+memcached_port = configuration.get('tcp-port')
+
+for memcached_unit in units:
+    command = 'service memcached status'
+    output, code = memcached_unit.run(command)
+
+    # open memcache to be able to connect from this machine
+    memcached_unit.run('ufw allow {}'.format(memcached_port))
+
+    if code != 0:
+        message = 'The ' + command + ' returned %d.' % code
+        print(output)
+        amulet.raise_status(amulet.FAIL, msg=message)
+    else:
+        message = 'The memcached service is running.'
+        print(output)
+        print(message)
+
+        memcached_address = memcached_unit.info['public-address']
+        try:
+            telnetlib.Telnet(memcached_address, memcached_port)
+        except TimeoutError:  # noqa this exception only available in py3
+            amulet.raise_status(amulet.FAIL, msg="Cannot connect to port %s" %
+                                memcached_port)
+
+primary, secondary = units
+
+# if primary points to secondary
+check_config(primary, secondary)
+
+# if secondary points to primary
+check_config(secondary, primary)
+
+# if replication primary -> secondary works
+check_replication(primary, secondary, memcached_port)
+
+# if replication secondary -> primary works
+check_replication(secondary, primary, memcached_port)

=== modified file 'unit_tests/test_memcached_hooks.py'
--- unit_tests/test_memcached_hooks.py	2015-02-04 12:16:33 +0000
+++ unit_tests/test_memcached_hooks.py	2015-02-18 02:12:32 +0000
@@ -1,5 +1,3 @@
-__author__ = 'Felipe Reyes <[e-mail]>'
-
 import mock
 import os
 import shutil
@@ -7,6 +5,8 @@
 from test_utils import CharmTestCase
 import memcached_hooks
 
+__author__ = 'Felipe Reyes <[e-mail]>'
+
 
 DOT = os.path.dirname(os.path.abspath(__file__))
 TO_PATCH = [
@@ -20,6 +20,8 @@
     'unit_get',
     'config',
     'log',
+    'oldest_peer',
+    'peer_units',
 ]
 FREE_MEM_SMALL = """             total       used       free     shared    \
 buffers     cached
@@ -52,9 +54,21 @@
         super(TestMemcachedHooks, self).tearDown()
         shutil.rmtree(self.tmpdir, ignore_errors=True)
 
+    @mock.patch('memcached_hooks.add_source')
     @mock.patch('subprocess.check_output')
-    def test_install(self, check_output):
+    def test_install(self, check_output, add_source):
+        configs = {
+            'repcached': True,
+            'repcached_origin': 'ppa:niedbalski/memcached-repcached'
+        }
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+
         memcached_hooks.install()
+        add_source.assert_called_with(configs['repcached_origin'])
 
         self.apt_update.assert_called_with(fatal=True)
         self.apt_install.assert_called_with(["memcached", "python-cheetah",
@@ -248,6 +262,7 @@
     @mock.patch('os.fchown')
     @mock.patch('os.chown')
     @mock.patch('memcached_utils.log')
+    @mock.patch('memcached_utils.dpkg_info_contains')
     @mock.patch('subprocess.Popen')
     @mock.patch('memcached_utils.config')
     @mock.patch('subprocess.check_output')
@@ -255,17 +270,227 @@
     @mock.patch('charmhelpers.contrib.network.ufw.service')
     @mock.patch('charmhelpers.core.hookenv.charm_dir')
     @mock.patch('charmhelpers.core.host.log')
-    def test_upgrade_charm(self, log, charm_dir, service, enable, check_output,
-                           config, popen, *args):
+    @mock.patch('memcached_hooks.add_source')
+    @mock.patch('memcached_hooks.config_changed')
+    def test_upgrade_charm(self, config_changed,
+                           add_source, log, charm_dir, service, enable,
+                           check_output, config, popen, dpkg, *args):
         p = mock.Mock()
         p.configure_mock(**{'communicate.return_value': ('stdout', 'stderr'),
                             'returncode': 0})
         popen.return_value = p
+        dpkg.return_value = True
 
         charm_dir.return_value = os.path.join(DOT, '..')
         config.return_value = 12111
+
+        configs = {
+            'repcached': True,
+            'repcached_origin': 'ppa:niedbalski/memcached-repcached'
+        }
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+
         memcached_hooks.upgrade_charm()
+
+        add_source.assert_called_with(configs['repcached_origin'])
         self.apt_install.assert_any_call(['memcached', 'python-cheetah',
                                           'python-memcache'], fatal=True)
         enable.assert_called_with()
         service.assert_called_with('ssh', 'open')
+
+    @mock.patch('replication.store_replica')
+    @mock.patch('memcached_hooks.replication.get_repcached_replica')
+    @mock.patch('replication.set_repcached_replica')
+    @mock.patch('memcached_hooks.config_changed')
+    def test_cluster_relation_joined_secondary(self, config_changed, set_rep,
+                                               get_rep, store):
+        configs = {'repcached': True}
+
+        def f(c):
+            return configs.get(c, None)
+
+        get_rep.return_value = ('10.0.0.1', '10.0.0.2')
+
+        self.config.side_effect = f
+        self.oldest_peer.return_value = False
+        memcached_hooks.cluster_relation_joined()
+        config_changed.assert_not_called()
+
+    @mock.patch('replication.get_repcached_replica')
+    @mock.patch('replication.set_repcached_replica')
+    @mock.patch('memcached_hooks.config_changed')
+    def test_cluster_relation_joined_master(self, config_changed, set_rep,
+                                            get_rep):
+        configs = {'repcached': True}
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+        self.peer_units.return_value = [0, 0]
+        self.oldest_peer.return_value = True
+        get_rep.return_value = ('10.0.0.1', '10.0.0.2')
+        memcached_hooks.cluster_relation_joined()
+
+        config_changed.assert_called_with(
+            replica='10.0.0.2')
+
+    @mock.patch('memcached_hooks.config_changed')
+    def test_cluster_relation_changed_master(self, config_changed):
+        configs = {'repcached': True}
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+        self.peer_units.return_value = [0, 0]
+        self.oldest_peer.return_value = True
+
+        memcached_hooks.cluster_relation_changed()
+        config_changed.assert_not_called()
+
+    @mock.patch('replication.store_replica')
+    @mock.patch('memcached_hooks.config_changed')
+    def test_cluster_relation_changed_secondary(self, config_changed, store):
+        configs = {'repcached': True}
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+
+        relations = {
+            'replica': '10.0.0.2',
+            'master': '10.0.0.1',
+        }
+
+        def r(k):
+            return relations.get(k, None)
+
+        self.relation_get.side_effect = r
+        self.unit_get.return_value = '10.0.0.2'
+
+        self.peer_units.return_value = [0, 0]
+        self.oldest_peer.return_value = False
+
+        memcached_hooks.cluster_relation_changed()
+        config_changed.assert_called_with(
+            replica=relations.get('master'))
+
+    @mock.patch('memcached_hooks.config_changed')
+    def test_cluster_relation_changed_secondary_fail(self, config_changed):
+        configs = {'repcached': True}
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+
+        relations = {
+            'replica': '10.0.0.2',
+            'master': '10.0.0.1',
+        }
+
+        def r(k):
+            return relations.get(k, None)
+
+        self.relation_get.side_effect = r
+        self.unit_get.return_value = '10.0.0.3'
+
+        self.peer_units.return_value = [0, 0]
+        self.oldest_peer.return_value = False
+
+        self.assertRaises(Exception, memcached_hooks.cluster_relation_changed)
+
+    @mock.patch('replication.get_current_replica')
+    @mock.patch('replication.unit_get')
+    @mock.patch('memcached_hooks.ufw.service')
+    @mock.patch('memcached_hooks.cache_relation_joined')
+    @mock.patch('charmhelpers.core.templating.render')
+    @mock.patch('replication.get_repcached_replica')
+    @mock.patch('memcached_utils.dpkg_info_contains')
+    def test_config_changed_replica(self, dpkg, get_replica, render,
+                                    cache_joined, ufw, unit_get, replica):
+        params = {
+            'tcp_port': None,
+            'disable_cas': None,
+            'mem_size': None,
+            'factor': None,
+            'connection_limit': None,
+            'udp_port': None,
+            'slab_page_size': None,
+            'threads': None,
+            'large_pages_enabled': False,
+            'min_item_size': None,
+            'repcached_port': None,
+            'request_limit': None,
+            'disable_auto_cleanup': None,
+        }
+
+        dpkg.return_value = True
+        replica.return_value = ('10.0.0.2')
+
+        configs = {
+            'repcached': True,
+            'memsize': 1,
+        }
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+        self.unit_get.return_value = '10.0.0.1'
+
+        memcached_hooks.config_changed()
+
+        params.update({'replica': '10.0.0.2'})
+        ufw.assert_called_with(params.get('repcached_port'), 'open')
+        render.assert_called_with('memcached.conf',
+                                  memcached_hooks.ETC_MEMCACHED_CONF,
+                                  params)
+
+    @mock.patch('memcached_hooks.cache_relation_joined')
+    @mock.patch('charmhelpers.contrib.network.ufw.service')
+    @mock.patch('memcached_hooks.replication.get_current_replica')
+    @mock.patch('charmhelpers.core.templating.render')
+    @mock.patch('memcached_utils.dpkg_info_contains')
+    def test_config_changed_no_replica(self, dpkg, render,
+                                       replica, ufw, cache):
+        params = {
+            'tcp_port': None,
+            'disable_cas': None,
+            'mem_size': None,
+            'factor': None,
+            'connection_limit': None,
+            'udp_port': None,
+            'slab_page_size': None,
+            'threads': None,
+            'replica': "10.0.0.2",
+            'repcached_port': None,
+            'large_pages_enabled': False,
+            'min_item_size': None,
+            'request_limit': None,
+            'disable_auto_cleanup': None,
+        }
+
+        replica.return_value = "10.0.0.2"
+        dpkg.return_value = True
+
+        configs = {
+            'repcached': True,
+            'memsize': 1,
+        }
+
+        def f(c):
+            return configs.get(c, None)
+
+        self.config.side_effect = f
+
+        memcached_hooks.config_changed()
+        render.assert_called_with('memcached.conf',
+                                  memcached_hooks.ETC_MEMCACHED_CONF,
+                                  params)