Initial commit moving openolt adapter from voltha-go to the new repo.
This version works with ponsim rather than openolt, this is temporary.
It is currently being fixed to work with openolt.
Change-Id: I34a800c98f050140b367e2d474b7aa8b79f34b9a
Signed-off-by: William Kurkian <wkurkian@cisco.com>
diff --git a/python/adapters/common/kvstore/etcd_client.py b/python/adapters/common/kvstore/etcd_client.py
new file mode 100644
index 0000000..e1850e7
--- /dev/null
+++ b/python/adapters/common/kvstore/etcd_client.py
@@ -0,0 +1,240 @@
+# Copyright 2018-present Open Networking Foundation
+#
+# 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.
+
+################################################################################
+#
+# Most of the txaioetcd methods provide a timeout parameter. This parameter
+# is likely intended to limit the amount of time spent by any one method
+# waiting for a response from the etcd server. However, if the server is
+# down, the method immediately throws a ConnectionRefusedError exception;
+# it does not perform any retries. The timeout parameter provided by the
+# methods in EtcdClient cover this contingency.
+#
+################################################################################
+
+from kv_client import DEFAULT_TIMEOUT, Event, KVClient, KVPair
+from structlog import get_logger
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
+from twisted.internet.error import ConnectionRefusedError
+from txaioetcd import Client, CompVersion, Failed, KeySet, OpGet, OpSet, Transaction
+
+log = get_logger()
+
+class EtcdClient(KVClient):
+
+ def __init__(self, kv_host, kv_port):
+ KVClient.__init__(self, kv_host, kv_port)
+ self.url = u'http://' + kv_host + u':' + str(kv_port)
+ self.client = Client(reactor, self.url)
+
+ @inlineCallbacks
+ def watch(self, key, key_change_callback, timeout=DEFAULT_TIMEOUT):
+ self.key_watches[key] = key_change_callback
+ result = yield self._op_with_retry('WATCH', key, None, timeout, callback=self.key_changed)
+ returnValue(result)
+
+ def key_changed(self, kv):
+ key = kv.key
+ value = kv.value
+ log.debug('key-changed', key=key, value=value)
+ # Notify client of key change event
+ if value is not None:
+ evt = Event(Event.PUT, key, value)
+ else:
+ evt = Event(Event.DELETE, key, None)
+ if key in self.key_watches:
+ self.key_watches[key](evt)
+
+ def close_watch(self, key, timeout=DEFAULT_TIMEOUT):
+ log.debug('close-watch', key=key)
+ if key in self.key_watches:
+ self.key_watches.pop(key)
+
+ @inlineCallbacks
+ def _op_with_retry(self, operation, key, value, timeout, *args, **kw):
+ log.debug('kv-op', operation=operation, key=key, timeout=timeout, args=args, kw=kw)
+ err = None
+ result = None
+ if type(key) == str:
+ key = bytes(key)
+ if value is not None:
+ value = bytes(value)
+ while True:
+ try:
+ if operation == 'GET':
+ result = yield self._get(key)
+ elif operation == 'LIST':
+ result, err = yield self._list(key)
+ elif operation == 'PUT':
+ # Put returns an object of type Revision
+ result = yield self.client.set(key, value, **kw)
+ elif operation == 'DELETE':
+ # Delete returns an object of type Deleted
+ result = yield self.client.delete(key)
+ elif operation == 'RESERVE':
+ result, err = yield self._reserve(key, value, **kw)
+ elif operation == 'RENEW':
+ result, err = yield self._renew_reservation(key)
+ elif operation == 'RELEASE':
+ result, err = yield self._release_reservation(key)
+ elif operation == 'RELEASE-ALL':
+ err = yield self._release_all_reservations()
+ elif operation == 'WATCH':
+ for name, val in kw.items():
+ if name == 'callback':
+ callback = val
+ break
+ result = self.client.watch([KeySet(key, prefix=True)], callback)
+ self._clear_backoff()
+ break
+ except ConnectionRefusedError as ex:
+ log.error('comms-exception', ex=ex)
+ yield self._backoff('etcd-not-up')
+ except Exception as ex:
+ log.error('etcd-exception', ex=ex)
+ err = ex
+
+ if timeout > 0 and self.retry_time > timeout:
+ err = 'operation-timed-out'
+ if err is not None:
+ self._clear_backoff()
+ break
+
+ returnValue((result, err))
+
+ @inlineCallbacks
+ def _get(self, key):
+ kvp = None
+ resp = yield self.client.get(key)
+ if resp.kvs is not None and len(resp.kvs) == 1:
+ kv = resp.kvs[0]
+ kvp = KVPair(kv.key, kv.value, kv.mod_revision)
+ returnValue(kvp)
+
+ @inlineCallbacks
+ def _list(self, key):
+ err = None
+ list = []
+ resp = yield self.client.get(KeySet(key, prefix=True))
+ if resp.kvs is not None and len(resp.kvs) > 0:
+ for kv in resp.kvs:
+ list.append(KVPair(kv.key, kv.value, kv.mod_revision))
+ returnValue((list, err))
+
+ @inlineCallbacks
+ def _reserve(self, key, value, **kw):
+ for name, val in kw.items():
+ if name == 'ttl':
+ ttl = val
+ break
+ reserved = False
+ err = 'reservation-failed'
+ owner = None
+
+ # Create a lease
+ lease = yield self.client.lease(ttl)
+
+ # Create a transaction
+ txn = Transaction(
+ compare=[ CompVersion(key, '==', 0) ],
+ success=[ OpSet(key, bytes(value), lease=lease) ],
+ failure=[ OpGet(key) ]
+ )
+ newly_acquired = False
+ try:
+ result = yield self.client.submit(txn)
+ except Failed as failed:
+ log.debug('key-already-present', key=key)
+ if len(failed.responses) > 0:
+ response = failed.responses[0]
+ if response.kvs is not None and len(response.kvs) > 0:
+ kv = response.kvs[0]
+ log.debug('key-already-present', value=kv.value)
+ if kv.value == value:
+ reserved = True
+ log.debug('key-already-reserved', key = kv.key, value=kv.value)
+ else:
+ newly_acquired = True
+ log.debug('key-was-absent', key=key, result=result)
+
+ # Check if reservation succeeded
+ resp = yield self.client.get(key)
+ if resp.kvs is not None and len(resp.kvs) == 1:
+ owner = resp.kvs[0].value
+ if owner == value:
+ if newly_acquired:
+ log.debug('key-reserved', key=key, value=value, ttl=ttl,
+ lease_id=lease.lease_id)
+ reserved = True
+ # Add key to reservation list
+ self.key_reservations[key] = lease
+ else:
+ log.debug("reservation-still-held")
+ else:
+ log.debug('reservation-held-by-another', value=owner)
+
+ if reserved:
+ err = None
+ returnValue((owner, err))
+
+ @inlineCallbacks
+ def _renew_reservation(self, key):
+ result = None
+ err = None
+ if key not in self.key_reservations:
+ err = 'key-not-reserved'
+ else:
+ lease = self.key_reservations[key]
+ # A successfully refreshed lease returns an object of type Header
+ result = yield lease.refresh()
+ if result is None:
+ err = 'lease-refresh-failed'
+ returnValue((result, err))
+
+ @inlineCallbacks
+ def _release_reservation(self, key):
+ err = None
+ if key not in self.key_reservations:
+ err = 'key-not-reserved'
+ else:
+ lease = self.key_reservations[key]
+ time_left = yield lease.remaining()
+ # A successfully revoked lease returns an object of type Header
+ log.debug('release-reservation', key=key, lease_id=lease.lease_id,
+ time_left_in_secs=time_left)
+ result = yield lease.revoke()
+ if result is None:
+ err = 'lease-revoke-failed'
+ self.key_reservations.pop(key)
+ returnValue((result, err))
+
+ @inlineCallbacks
+ def _release_all_reservations(self):
+ err = None
+ keys_to_delete = []
+ for key in self.key_reservations:
+ lease = self.key_reservations[key]
+ time_left = yield lease.remaining()
+ # A successfully revoked lease returns an object of type Header
+ log.debug('release-reservation', key=key, lease_id=lease.lease_id,
+ time_left_in_secs=time_left)
+ result = yield lease.revoke()
+ if result is None:
+ err = 'lease-revoke-failed'
+ log.debug('lease-revoke', result=result)
+ keys_to_delete.append(key)
+ for key in keys_to_delete:
+ self.key_reservations.pop(key)
+ returnValue(err)