1#!/usr/bin/env python3
2#
3#   Copyright 2019 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import logging
18import time
19
20from threading import Thread
21
22from acts.libs.logging import log_stream
23from acts.libs.logging.log_stream import LogStyles
24from acts.controllers.android_lib.logcat import TimestampTracker
25from acts.controllers.fuchsia_lib.utils_lib import create_ssh_connection
26
27
28def _log_line_func(log, timestamp_tracker):
29    """Returns a lambda that logs a message to the given logger."""
30
31    def log_line(message):
32        timestamp_tracker.read_output(message)
33        log.info(message)
34
35    return log_line
36
37
38def start_syslog(serial,
39                 base_path,
40                 ip_address,
41                 ssh_username,
42                 ssh_config,
43                 extra_params=''):
44    """Creates a FuchsiaSyslogProcess that automatically attempts to reconnect.
45
46    Args:
47        serial: The unique identifier for the device.
48        base_path: The base directory used for syslog file output.
49        ip_address: The ip address of the device to get the syslog.
50        ssh_username: Username for the device for the Fuchsia Device.
51        ssh_config: Location of the ssh_config for connecting to the remote
52            device
53        extra_params: Any additional params to be added to the syslog cmdline.
54
55    Returns:
56        A FuchsiaSyslogProcess object.
57    """
58    logger = log_stream.create_logger(
59        'fuchsia_log_%s' % serial, base_path=base_path,
60        log_styles=(LogStyles.LOG_DEBUG | LogStyles.MONOLITH_LOG))
61    syslog = FuchsiaSyslogProcess(ssh_username,
62                                  ssh_config,
63                                  ip_address,
64                                  extra_params)
65    timestamp_tracker = TimestampTracker()
66    syslog.set_on_output_callback(_log_line_func(logger, timestamp_tracker))
67    return syslog
68
69
70class FuchsiaSyslogError(Exception):
71    """Raised when invalid operations are run on a Fuchsia Syslog."""
72
73
74class FuchsiaSyslogProcess(object):
75    """A class representing a Fuchsia Syslog object that communicates over ssh.
76    """
77
78    def __init__(self, ssh_username, ssh_config, ip_address, extra_params):
79        """
80        Args:
81            ssh_username: The username to connect to Fuchsia over ssh.
82            ssh_config: The ssh config that holds the information to connect to
83            a Fuchsia device over ssh.
84            ip_address: The ip address of the Fuchsia device.
85        """
86        self.ssh_config = ssh_config
87        self.ip_address = ip_address
88        self.extra_params = extra_params
89        self.ssh_username = ssh_username
90        self._output_file = None
91        self._ssh_client = None
92        self._listening_thread = None
93        self._redirection_thread = None
94        self._on_output_callback = lambda *args, **kw: None
95
96        self._started = False
97        self._stopped = False
98
99    def start(self):
100        """Starts reading the data from the syslog ssh connection."""
101        if self._started:
102            raise FuchsiaSyslogError('Syslog has already started for '
103                                     'FuchsiaDevice (%s).' % self.ip_address)
104        self._started = True
105
106        self._listening_thread = Thread(target=self._exec_loop)
107        self._listening_thread.start()
108
109        time_up_at = time.time() + 10
110
111        while self._ssh_client is None:
112            if time.time() > time_up_at:
113                raise FuchsiaSyslogError('Unable to connect to syslog!')
114
115        self._stopped = False
116
117    def stop(self):
118        """Stops listening to the syslog ssh connection and coalesces the
119        threads.
120        """
121        if self._stopped:
122            raise FuchsiaSyslogError('Syslog is already being stopped for '
123                                     'FuchsiaDevice (%s).' % self.ip_address)
124        self._stopped = True
125
126        try:
127            self._ssh_client.close()
128        except Exception as e:
129            raise e
130        finally:
131            self._join_threads()
132            self._started = False
133            return None
134
135    def _join_threads(self):
136        """Waits for the threads associated with the process to terminate."""
137        if self._listening_thread is not None:
138            self._listening_thread.join()
139            self._listening_thread = None
140
141        if self._redirection_thread is not None:
142            self._redirection_thread.join()
143            self._redirection_thread = None
144
145    def _redirect_output(self):
146        """Redirects the output from the ssh connection into the
147        on_output_callback.
148        """
149        while True:
150            line = self._output_file.readline()
151
152            if not line:
153                return
154            else:
155                # Output the line without trailing \n and whitespace.
156                self._on_output_callback(line.rstrip())
157
158    def set_on_output_callback(self, on_output_callback, binary=False):
159        """Sets the on_output_callback function.
160
161        Args:
162            on_output_callback: The function to be called when output is sent to
163                the output. The output callback has the following signature:
164
165                >>> def on_output_callback(output_line):
166                >>>     return None
167
168            binary: If True, read the process output as raw binary.
169        Returns:
170            self
171        """
172        self._on_output_callback = on_output_callback
173        self._binary_output = binary
174        return self
175
176    def __start_process(self):
177        """A convenient wrapper function for starting the ssh connection and
178        starting the syslog."""
179
180        self._ssh_client = create_ssh_connection(self.ip_address,
181                                                 self.ssh_username,
182                                                 self.ssh_config)
183        transport = self._ssh_client.get_transport()
184        channel = transport.open_session()
185        channel.get_pty()
186        self._output_file = channel.makefile()
187        logging.debug('Starting FuchsiaDevice (%s) syslog over ssh.'
188                      % self.ssh_username)
189        channel.exec_command('log_listener %s' % self.extra_params)
190        return transport
191
192    def _exec_loop(self):
193        """Executes a ssh connection to the Fuchsia Device syslog in a loop.
194
195        When the ssh connection terminates without stop() being called,
196        the threads are coalesced and the syslog is restarted.
197        """
198        start_up = True
199        while True:
200            if self._stopped:
201                break
202            else:
203                if start_up or not ssh_transport.is_alive():
204                    if start_up:
205                        logging.debug('Starting SSH connection for '
206                                      'FuchsiaDevice (%s) syslog.'
207                                      % self.ip_address)
208                        start_up = False
209                    else:
210                        logging.debug('SSH connection for FuchsiaDevice (%s) is'
211                                      ' down.  Restarting.' % self.ip_address)
212                    ssh_transport = self.__start_process()
213                    self._redirection_thread = Thread(
214                        target=self._redirect_output)
215                    self._redirection_thread.start()
216                    self._redirection_thread.join()
217