Clear table_map if RotateEvent has timestamp of 0 (julien-duponchelle#197)

troyjfarrell · baloo · commit 084712bf8cf9 · 2017-04-02T17:27:37.000Z
* Clear table_map if RotateEvent has timestamp of 0

MySQL generates two RotateEvents when flushing binary logs.  The first
is added to the end of the old binary log.  This RotateEvent has a valid
timestamp.  The second is added to the beginning of the new binary log.
This RotateEvent has a timestamp of 0.

Under some circumstances, e.g., the MySQL server crashes, the first
RotateEvent will not be written to the binary log.  (See the MySQL
Internals Manual 20.9.)

Before this change, when instantiating BinLogStreamReader with a
positive skip_to_timestamp argument, the RotateEvents with a timestamp
of 0 were ignored.  This could cause silent corruption of the table_map.

This change ensures that RotateEvents always reset the table_map,
log_pos and log_file attributes.

* Update documentation on running tests

* Run the test_no_trailing_rotate_event with sudo

This change alters the Travis CI configuration to run the
test_no_trailing_rotate_event with sudo and exclude it from the normal
test run.

* Fix sudo not finding the nosetests command
diff --git a/.travis.yml b/.travis.yml
@@ -22,5 +22,7 @@ before_script:
   - env | grep DB
   - bash -c "if [ '$DB' = 'mysql57' ]; then sudo ./scripts/install_mysql57.sh; fi"
   - bash -c "if [ '$DB' = 'mysql56' ]; then sudo ./scripts/install_mysql56.sh; fi"
+  - which nosetests > ~/where-is-nosetests
 script:
-  - "nosetests"
+  - "sudo $(cat ~/where-is-nosetests) pymysqlreplication.tests.test_abnormal:TestAbnormalBinLogStreamReader.test_no_trailing_rotate_event"
+  - "nosetests -e test_no_trailing_rotate_event"
diff --git a/docs/developement.rst b/docs/developement.rst
@@ -37,11 +37,14 @@ Make sure you have the following configuration set in your mysql config file (us
     enforce_gtid_consistency
 
 
-To run tests:
+Run tests with
 
 ::
 
-    python setup.py test
+    py.test -k "not test_no_trailing_rotate_event"
+
+This will skip the ``test_no_trailing_rotate_event`` which requires that the
+user running the test have permission to alter the binary log files.
 
 Running mysql in docker (main):
 
diff --git a/pymysqlreplication/binlogstream.py b/pymysqlreplication/binlogstream.py
@@ -431,14 +431,6 @@ def fetchone(self):
                                                self.__freeze_schema,
                                                self.__fail_on_table_metadata_unavailable)
 
-            if self.skip_to_timestamp and binlog_event.timestamp < self.skip_to_timestamp:
-                continue
-
-            if binlog_event.event_type == TABLE_MAP_EVENT and \
-                    binlog_event.event is not None:
-                self.table_map[binlog_event.event.table_id] = \
-                    binlog_event.event.get_table()
-
             if binlog_event.event_type == ROTATE_EVENT:
                 self.log_pos = binlog_event.event.position
                 self.log_file = binlog_event.event.next_binlog
@@ -455,6 +447,38 @@ def fetchone(self):
             elif binlog_event.log_pos:
                 self.log_pos = binlog_event.log_pos
 
+            # This check must not occur before clearing the ``table_map`` as a
+            # result of a RotateEvent.
+            #
+            # The first RotateEvent in a binlog file has a timestamp of
+            # zero.  If the server has moved to a new log and not written a
+            # timestamped RotateEvent at the end of the previous log, the
+            # RotateEvent at the beginning of the new log will be ignored
+            # if the caller provided a positive ``skip_to_timestamp``
+            # value.  This will result in the ``table_map`` becoming
+            # corrupt.
+            #
+            # https://dev.mysql.com/doc/internals/en/event-data-for-specific-event-types.html
+            # From the MySQL Internals Manual:
+            #
+            #   ROTATE_EVENT is generated locally and written to the binary
+            #   log on the master. It is written to the relay log on the
+            #   slave when FLUSH LOGS occurs, and when receiving a
+            #   ROTATE_EVENT from the master. In the latter case, there
+            #   will be two rotate events in total originating on different
+            #   servers.
+            #
+            #   There are conditions under which the terminating
+            #   log-rotation event does not occur. For example, the server
+            #   might crash.
+            if self.skip_to_timestamp and binlog_event.timestamp < self.skip_to_timestamp:
+                continue
+
+            if binlog_event.event_type == TABLE_MAP_EVENT and \
+                    binlog_event.event is not None:
+                self.table_map[binlog_event.event.table_id] = \
+                    binlog_event.event.get_table()
+
             # event is none if we have filter it on packet level
             # we filter also not allowed events
             if binlog_event.event is None or (binlog_event.event.__class__ not in self.__allowed_events):
diff --git a/pymysqlreplication/tests/binlogfilereader.py b/pymysqlreplication/tests/binlogfilereader.py
@@ -0,0 +1,150 @@
+'''Read binlog files'''
+import struct
+
+from pymysql.util import byte2int
+from pymysqlreplication import constants
+from pymysqlreplication.event import FormatDescriptionEvent
+from pymysqlreplication.event import QueryEvent
+from pymysqlreplication.event import RotateEvent
+from pymysqlreplication.event import XidEvent
+from pymysqlreplication.row_event import TableMapEvent
+from pymysqlreplication.row_event import WriteRowsEvent
+
+class SimpleBinLogFileReader(object):
+    '''Read binlog files'''
+
+    _expected_magic = b'\xfebin'
+
+    def __init__(self, file_path, only_events=None):
+        self._current_event = None
+        self._file = None
+        self._file_path = file_path
+        self._only_events = only_events
+        self._pos = None
+
+    def fetchone(self):
+        '''Fetch one record from the binlog file'''
+        if self._pos is None or self._pos < 4:
+            self._read_magic()
+        while True:
+            event = self._read_event()
+            self._current_event = event
+            if event is None:
+                return None
+            if self._filter_events(event):
+                return event
+
+    def truncatebinlog(self):
+        '''Truncate the binlog file at the current event'''
+        if self._current_event is not None:
+            self._file.truncate(self._current_event.pos)
+
+    def _filter_events(self, event):
+        '''Return True if an event can be returned'''
+        # It would be good if we could reuse the __event_map in
+        # packet.BinLogPacketWrapper.
+        event_type = {
+            constants.QUERY_EVENT: QueryEvent,
+            constants.ROTATE_EVENT: RotateEvent,
+            constants.FORMAT_DESCRIPTION_EVENT: FormatDescriptionEvent,
+            constants.XID_EVENT: XidEvent,
+            constants.TABLE_MAP_EVENT: TableMapEvent,
+            constants.WRITE_ROWS_EVENT_V2: WriteRowsEvent,
+        }.get(event.event_type)
+        return event_type in self._only_events
+
+    def _open_file(self):
+        '''Open the file at ``self._file_path``'''
+        if self._file is None:
+            self._file = open(self._file_path, 'rb+')
+            self._pos = self._file.tell()
+            assert self._pos == 0
+
+    def _read_event(self):
+        '''Read an event from the binlog file'''
+        # Assuming a binlog version > 1
+        headerlength = 19
+        header = self._file.read(headerlength)
+        event_pos = self._pos
+        self._pos += len(header)
+        if len(header) == 0:
+            return None
+        event = SimpleBinLogEvent(header)
+        event.set_pos(event_pos)
+        if event.event_size < headerlength:
+            messagefmt = 'Event size {0} is too small'
+            message = messagefmt.format(event.event_size)
+            raise EventSizeTooSmallError(message)
+        else:
+            body = self._file.read(event.event_size - headerlength)
+            self._pos += len(body)
+            event.set_body(body)
+        return event
+
+    def _read_magic(self):
+        '''Read the first four *magic* bytes of the binlog file'''
+        self._open_file()
+        if self._pos == 0:
+            magic = self._file.read(4)
+            if magic == self._expected_magic:
+                self._pos += len(magic)
+            else:
+                messagefmt = 'Magic bytes {0!r} did not match expected {1!r}'
+                message = messagefmt.format(magic, self._expected_magic)
+                raise BadMagicBytesError(message)
+
+    def __iter__(self):
+        return iter(self.fetchone, None)
+
+    def __repr__(self):
+        cls = self.__class__
+        mod = cls.__module__
+        name = cls.__name__
+        only = [type(x).__name__ for x in self._only_events]
+        fmt = '<{mod}.{name}(file_path={fpath}, only_events={only})>'
+        return fmt.format(mod=mod, name=name, fpath=self._file_path, only=only)
+
+
+# pylint: disable=too-many-instance-attributes
+class SimpleBinLogEvent(object):
+    '''An event from a binlog file'''
+
+    def __init__(self, header):
+        '''Initialize the Event with the event header'''
+        unpacked = struct.unpack('<IcIIIH', header)
+        self.timestamp = unpacked[0]
+        self.event_type = byte2int(unpacked[1])
+        self.server_id = unpacked[2]
+        self.event_size = unpacked[3]
+        self.log_pos = unpacked[4]
+        self.flags = unpacked[5]
+
+        self.body = None
+        self.pos = None
+
+    def set_body(self, body):
+        '''Save the body bytes'''
+        self.body = body
+
+    def set_pos(self, pos):
+        '''Save the event position'''
+        self.pos = pos
+
+    def __repr__(self):
+        cls = self.__class__
+        mod = cls.__module__
+        name = cls.__name__
+        fmt = '<{mod}.{name}(timestamp={ts}, event_type={et}, log_pos={pos})>'
+        return fmt.format(
+            mod=mod,
+            name=name,
+            ts=int(self.timestamp),
+            et=self.event_type,
+            pos=self.log_pos)
+
+
+class BadMagicBytesError(Exception):
+    '''The binlog file magic bytes did not match the specification'''
+
+class EventSizeTooSmallError(Exception):
+    '''The event size was smaller than the length of the event header'''
diff --git a/pymysqlreplication/tests/test_abnormal.py b/pymysqlreplication/tests/test_abnormal.py
@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+'''Test abnormal conditions, such as caused by a MySQL crash
+'''
+import os.path
+
+from pymysqlreplication.tests import base
+from pymysqlreplication.tests.binlogfilereader import SimpleBinLogFileReader
+from pymysqlreplication import BinLogStreamReader
+from pymysqlreplication.event import GtidEvent
+from pymysqlreplication.event import RotateEvent
+
+class TestAbnormalBinLogStreamReader(base.PyMySQLReplicationTestCase):
+    '''Test abnormal condition handling in the BinLogStreamReader
+    '''
+
+    @staticmethod
+    def ignored_events():
+        '''Events the BinLogStreamReader should ignore'''
+        return [GtidEvent]
+
+    def test_no_trailing_rotate_event(self):
+        '''A missing RotateEvent and skip_to_timestamp cause corruption
+
+        This test shows that a binlog file which lacks the trailing RotateEvent
+        and the use of the ``skip_to_timestamp`` argument together can cause
+        the table_map to become corrupt.  The trailing RotateEvent has a
+        timestamp, but may be lost if the server crashes.  The leading
+        RotateEvent in the next binlog file always has a timestamp of 0, thus
+        is discarded when ``skip_to_timestamp`` is greater than zero.
+        '''
+        self.execute(
+            'CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, '
+            'data VARCHAR (50) NOT NULL, PRIMARY KEY(id))')
+        self.execute('SET AUTOCOMMIT = 0')
+        self.execute('INSERT INTO test(id, data) VALUES (1, "Hello")')
+        self.execute('COMMIT')
+        timestamp = self.execute('SELECT UNIX_TIMESTAMP()').fetchone()[0]
+        self.execute('FLUSH BINARY LOGS')
+        self.execute('INSERT INTO test(id, data) VALUES (2, "Hi")')
+        self.stream.close()
+        self._remove_trailing_rotate_event_from_first_binlog()
+
+        binlog = self.execute("SHOW BINARY LOGS").fetchone()[0]
+
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            log_pos=4,
+            log_file=binlog,
+            skip_to_timestamp=timestamp,
+            ignored_events=self.ignored_events())
+        for _ in self.stream:
+            pass
+        # The table_map should be empty because of the binlog being rotated.
+        self.assertEqual({}, self.stream.table_map)
+
+    def _remove_trailing_rotate_event_from_first_binlog(self):
+        '''Remove the trailing RotateEvent from the first binlog
+
+        According to the MySQL Internals Manual, a RotateEvent will be added to
+        the end of a binlog when the binlog is rotated.  This may not happen if
+        the server crashes, for example.
+
+        This method removes the trailing RotateEvent to verify that the library
+        properly handles this case.
+        '''
+        datadir = self.execute("SHOW VARIABLES LIKE 'datadir'").fetchone()[1]
+        binlog = self.execute("SHOW BINARY LOGS").fetchone()[0]
+        binlogpath = os.path.join(datadir, binlog)
+
+        reader = SimpleBinLogFileReader(binlogpath, only_events=[RotateEvent])
+        for _ in reader:
+            reader.truncatebinlog()
+            break
