← Back to branch summary

~juju-deployers/juju-deployer/darwin

« back to all changes in this revision

Viewing changes to deployer/env/go.py

  • Committer: Adam Gandelman
  • Date: 2013-09-03 20:44:14 UTC
  • mfrom: (114 darwin)
  • mto: This revision was merged to the branch mainline in revision 119.
  • Revision ID: adamg@canonical.com-20130903204414-xsqqz2gp83dp5d2o
Merge lp:juju-deployer/darwin.

Show diffs side-by-side

added added

removed removed

Lines of Context:
deployer/env/go.py
 
1
import errno
 
2
import socket
 
3
import time
 
4
 
 
5
from .base import BaseEnvironment
 
6
from ..utils import ErrorExit
 
7
 
 
8
from jujuclient import Environment as EnvironmentClient, UnitErrors, EnvError
 
9
 
 
10
 
 
11
class GoEnvironment(BaseEnvironment):
 
12
 
 
13
    def __init__(self, name, options=None, endpoint=None):
 
14
        self.name = name
 
15
        self.options = options
 
16
        self.api_endpoint = endpoint
 
17
        self.client = None
 
18
 
 
19
    def _get_token(self):
 
20
        config = self._get_env_config()
 
21
        return config['admin-secret']
 
22
 
 
23
    def add_units(self, service_name, num_units):
 
24
        return self.client.add_units(service_name, num_units)
 
25
 
 
26
    def add_relation(self, endpoint_a, endpoint_b):
 
27
        return self.client.add_relation(endpoint_a, endpoint_b)
 
28
 
 
29
    def close(self):
 
30
        if self.client:
 
31
            self.client.close()
 
32
 
 
33
    def connect(self):
 
34
        if not self.api_endpoint:
 
35
            # Should really have a cheaper/faster way of getting the endpoint
 
36
            # ala juju status 0 for a get_endpoint method.
 
37
            self.get_cli_status()
 
38
        while True:
 
39
            try:
 
40
                self.client = EnvironmentClient(self.api_endpoint)
 
41
            except socket.error, e:
 
42
                if e.errno != errno.ETIMEDOUT:
 
43
                    raise
 
44
                continue
 
45
            else:
 
46
                break
 
47
        self.client.login(self._get_token())
 
48
        self.log.debug("Connected to environment")
 
49
 
 
50
    def get_config(self, svc_name):
 
51
        return self.client.get_config(svc_name)
 
52
 
 
53
    def get_constraints(self, svc_name):
 
54
        return self.client.get_constraints(svc_name)
 
55
 
 
56
    def get_cli_status(self):
 
57
        status = super(GoEnvironment, self).get_cli_status()
 
58
        # Opportunistic, see connect method comment.
 
59
        if not self.api_endpoint:
 
60
            self.api_endpoint = "wss://%s:17070/" % (
 
61
                status["machines"]["0"]["dns-name"])
 
62
        return status
 
63
 
 
64
    def reset(self,
 
65
              terminate_machines=False,
 
66
              terminate_delay=0,
 
67
              timeout=360, watch=False):
 
68
        """Destroy/reset the environment."""
 
69
        status = self.status()
 
70
        destroyed = False
 
71
        for s in status.get('services', {}).keys():
 
72
            self.log.debug(" Destroying service %s", s)
 
73
            self.client.destroy_service(s)
 
74
            destroyed = True
 
75
 
 
76
        if destroyed:
 
77
            # Mark any errors as resolved so destruction can proceed.
 
78
            self.resolve_errors()
 
79
 
 
80
            # Wait for units
 
81
            self.wait_for_units(timeout, "removed", watch=watch)
 
82
 
 
83
        # The only value to not terminating is keeping the data on the
 
84
        # machines around.
 
85
        if not terminate_machines:
 
86
            self.log.info(
 
87
                " warning: juju-core machines are not reusable for units")
 
88
            return
 
89
        self._terminate_machines(status, watch, terminate_delay)
 
90
 
 
91
    def _terminate_machines(self, status, watch, terminate_wait):
 
92
        """Terminate all machines, optionally wait for termination.
 
93
        """
 
94
        # Terminate machines
 
95
        self.log.debug(" Terminating machines")
 
96
 
 
97
        # Don't bother if there are no service unit machines
 
98
        if len(status['machines']) == 1:
 
99
            return
 
100
 
 
101
        for mid in status['machines'].keys():
 
102
            if mid == "0":
 
103
                continue
 
104
            self.log.debug("  Terminating machine %s", mid)
 
105
            self.terminate_machine(mid)
 
106
 
 
107
        if terminate_wait:
 
108
            self.log.info("  Waiting for machine termination")
 
109
            callback = watch and self._delta_event_log or None
 
110
            self.client.wait_for_no_machines(None, callback)
 
111
 
 
112
    def _check_timeout(self, etime):
 
113
        w_timeout = etime - time.time()
 
114
        if w_timeout < 0:
 
115
            self.log.error("Timeout reached while resolving errors")
 
116
            raise ErrorExit()
 
117
        return w_timeout
 
118
 
 
119
    def resolve_errors(self, retry_count=0, timeout=600, watch=False, delay=5):
 
120
        """Resolve any unit errors in the environment.
 
121
 
 
122
        If retry_count is given then the hook execution is reattempted. The
 
123
        system will do up to retry_count passes through the system resolving
 
124
        errors.
 
125
 
 
126
        If retry count is not given, the error is marked resolved permanently.
 
127
        """
 
128
        etime = time.time() + timeout
 
129
        count = 0
 
130
        while True:
 
131
            error_units = self._get_units_in_error()
 
132
            for e_uid in error_units:
 
133
                try:
 
134
                    self.client.resolved(e_uid, retry=bool(retry_count))
 
135
                    self.log.debug("  Resolving error on %s", e_uid)
 
136
                except EnvError, e:
 
137
                    if 'already resolved' in e:
 
138
                        continue
 
139
 
 
140
            if not error_units and count:
 
141
                if not count:
 
142
                    self.log.debug("  No unit errors found.")
 
143
                else:
 
144
                    self.log.debug("  No more unit errors found.")
 
145
                return
 
146
 
 
147
            w_timeout = self._check_timeout(etime)
 
148
            if retry_count:
 
149
                time.sleep(delay)
 
150
 
 
151
            count += 1
 
152
            try:
 
153
                self.wait_for_units(
 
154
                    timeout=int(w_timeout), watch=True, no_exit=True)
 
155
            except UnitErrors, e:
 
156
                if retry_count == count:
 
157
                    self.log.info(
 
158
                        " Retry count %d exhausted, but units in error (%s)",
 
159
                        retry_count, " ".join(u['Name'] for u in e.errors))
 
160
                    return
 
161
            else:
 
162
                return
 
163
 
 
164
    def status(self):
 
165
        return self.client.get_stat()
 
166
 
 
167
    def wait_for_units(
 
168
            self, timeout, goal_state="started", watch=False, no_exit=False):
 
169
        """Wait for units to reach a given condition.
 
170
        """
 
171
        callback = watch and self._delta_event_log or None
 
172
        try:
 
173
            self.client.wait_for_units(timeout, goal_state, callback=callback)
 
174
        except UnitErrors, e:
 
175
            error_units = [
 
176
                "unit: %s: machine: %s agent-state: %s details: %s" % (
 
177
                    u['Name'], u['MachineId'], u['Status'], u['StatusInfo']
 
178
                )
 
179
                for u in e.errors]
 
180
            if no_exit:
 
181
                raise
 
182
            self.log.error("The following units had errors:\n   %s" % (
 
183
                "   \n".join(error_units)))
 
184
            raise ErrorExit()
 
185
 
 
186
    def _delta_event_log(self, et, ct, d):
 
187
        # event type, change type, data
 
188
        name = d.get('Name', d.get('Id', 'unknown'))
 
189
        state = d.get('Status', d.get('Life', 'unknown'))
 
190
        if et == "relation":
 
191
            name = self._format_endpoints(d['Endpoints'])
 
192
            state = "created"
 
193
            if ct == "remove":
 
194
                state = "removed"
 
195
        self.log.debug(
 
196
            " Delta %s: %s %s:%s", et, name, ct, state)
 
197
 
 
198
    def _format_endpoints(self, eps):
 
199
        if len(eps) == 1:
 
200
            ep = eps.pop()
 
201
            return "[%s:%s:%s]" % (
 
202
                ep['ServiceName'],
 
203
                ep['Relation']['Name'],
 
204
                ep['Relation']['Role'])
 
205
 
 
206
        return "[%s:%s <-> %s:%s]" % (
 
207
            eps[0]['ServiceName'],
 
208
            eps[0]['Relation']['Name'],
 
209
            eps[1]['ServiceName'],
 
210
            eps[1]['Relation']['Name'])