binlog.c

/*
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program 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 General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.

        Authors: 	Domas Mituzas, Facebook ( domas at fb dot com )
                        Mark Leith, Oracle Corporation (mark dot leith at oracle
   dot com) Andrew Hutchings, SkySQL (andrew at skysql dot com)

*/

#include <glib.h>
#include <glib/gstdio.h>
#include <my_global.h>
#include <mysql.h>
#include <my_sys.h>
#include <mysqld_error.h>
#include <sql_common.h>
#include <string.h>
#include <zlib.h>
#include "mydumper.h"
#include "binlog.h"

#define BINLOG_MAGIC "\xfe\x62\x69\x6e"

#define EVENT_HEADER_LENGTH 19
#define EVENT_ROTATE_FIXED_LENGTH 8

enum event_postions {
  EVENT_TIMESTAMP_POSITION = 0,
  EVENT_TYPE_POSITION = 4,
  EVENT_SERVERID_POSITION = 5,
  EVENT_LENGTH_POSITION = 9,
  EVENT_NEXT_POSITION = 13,
  EVENT_FLAGS_POSITION = 17,
  EVENT_EXTRA_FLAGS_POSITION =
      19 // currently unused in v4 binlogs, but a good marker for end of header
};

enum event_type {
  ROTATE_EVENT = 4,
  FORMAT_DESCRIPTION_EVENT = 15,
  EVENT_TOO_SHORT = 254 // arbitrary high number, in 5.1 the max event type
                        // number is 27 so this should be fine for a while
};

extern int compress_output;
extern gboolean daemon_mode;
extern gboolean shutdown_triggered;

FILE *new_binlog_file(char *binlog_file, const char *binlog_dir);
void close_binlog_file(FILE *outfile);
char *rotate_file_name(const char *buf);

void get_binlogs(MYSQL *conn, struct configuration *conf) {
  // TODO: find logs we already have, use start position based on position of
  // last log.
  MYSQL_RES *result;
  MYSQL_ROW row;
  char *last_filename = NULL;
  guint64 last_position;

  // Only snapshot dump the binlogs once in daemon mode
  static gboolean got_binlogs = FALSE;
  if (got_binlogs)
    return;
  else
    got_binlogs = TRUE;

  if (mysql_query(conn, "SHOW MASTER STATUS")) {
    g_critical("Error: Could not execute query: %s", mysql_error(conn));
    return;
  }

  result = mysql_store_result(conn);
  if ((row = mysql_fetch_row(result))) {
    last_filename = g_strdup(row[0]);
    last_position = strtoll(row[1], NULL, 10);
  } else {
    g_critical("Error: Could not obtain binary log stop position");
    if (last_filename != NULL)
      g_free(last_filename);
    return;
  }
  mysql_free_result(result);

  if (mysql_query(conn, "SHOW BINARY LOGS")) {
    g_critical("Error: Could not execute query: %s", mysql_error(conn));
    if (last_filename != NULL)
      g_free(last_filename);
    return;
  }

  result = mysql_store_result(conn);
  while ((row = mysql_fetch_row(result))) {
    struct job *j = g_new0(struct job, 1);
    struct binlog_job *bj = g_new0(struct binlog_job, 1);
    j->job_data = (void *)bj;
    bj->filename = g_strdup(row[0]);
    bj->start_position = 4;
    bj->stop_position =
        (!strcasecmp(row[0], last_filename)) ? last_position : 0;
    j->conf = conf;
    j->type = JOB_BINLOG;
    g_async_queue_push(conf->queue, j);
  }
  mysql_free_result(result);
  if (last_filename != NULL)
    g_free(last_filename);
}

void get_binlog_file(MYSQL *conn, char *binlog_file,
                     const char *binlog_directory, guint64 start_position,
                     guint64 stop_position, gboolean continuous) {
  // set serverID = max serverID - threadID to try an eliminate conflicts,
  // 0 is bad because mysqld will disconnect at the end of the last log
  // dupes aren't too bad since it is up to the client to check for them
  uchar buf[128];
  // We need to read the raw network packets
  NET *net;
  net = &conn->net;
  unsigned long len;
  FILE *outfile;
  guint32 event_type;
  gboolean read_error = FALSE;
  gboolean read_end = FALSE;
  gboolean rotated = FALSE;
  guint32 server_id = G_MAXUINT32 - mysql_thread_id(conn);
  guint64 pos_counter = 0;

  int4store(buf, (guint32)start_position);
  // Binlog flags (2 byte int)
  int2store(buf + 4, 0);
  // ServerID
  int4store(buf + 6, server_id);
  memcpy(buf + 10, binlog_file, strlen(binlog_file));
#if MYSQL_VERSION_ID < 50100
  if (simple_command(conn, COM_BINLOG_DUMP, (const char *)buf,
#else
  if (simple_command(conn, COM_BINLOG_DUMP, buf,
#endif
                     strlen(binlog_file) + 10, 1)) {
    g_critical("Error: binlog: Critical error whilst requesting binary log");
  }

  while (1) {
    outfile = new_binlog_file(binlog_file, binlog_directory);
    if (outfile == NULL) {
      g_critical("Error: binlog: Could not create binlog file '%s', %d",
                 binlog_file, errno);
      return;
    }

    write_binlog(outfile, BINLOG_MAGIC, 4);
    while (1) {
      len = 0;
      if (net->vio != 0)
        len = my_net_read(net);
      if ((len == 0) || (len == ~(unsigned long)0)) {
        // Net timeout (set to 1 second)
        if (mysql_errno(conn) == ER_NET_READ_INTERRUPTED) {
          if (shutdown_triggered) {
            close_binlog_file(outfile);
            return;
          } else {
            continue;
          }
          // A real error
        } else {
          g_critical("Error: binlog: Network packet read error getting binlog "
                     "file: %s",
                     binlog_file);
          close_binlog_file(outfile);
          return;
        }
      }
      if (len < 8 && net->read_pos[0]) {
        // end of data
        break;
      }
      pos_counter += len;
      event_type = get_event((const char *)net->read_pos + 1, len - 1);
      switch (event_type) {
      case EVENT_TOO_SHORT:
        g_critical("Error: binlog: Event too short in binlog file: %s",
                   binlog_file);
        read_error = TRUE;
        break;
      case ROTATE_EVENT:
        if (rotated) {
          read_end = TRUE;
        } else {
          len = 1;
          rotated = TRUE;
        }
        break;
      default:
        // if we get this far this is a normal event to record
        break;
      }
      if (read_error)
        break;
      write_binlog(outfile, (const char *)net->read_pos + 1, len - 1);
      if (read_end) {
        if (!continuous) {
          break;
        } else {
          g_free(binlog_file);
          binlog_file = rotate_file_name((const char *)net->read_pos + 1);
          break;
        }
      }
      // stop if we are at requested end of last log
      if ((stop_position > 0) && (pos_counter >= stop_position))
        break;
    }
    close_binlog_file(outfile);
    if ((!continuous) || (!read_end))
      break;

    if (continuous && read_end) {
      read_end = FALSE;
      rotated = FALSE;
    }
  }
}

char *rotate_file_name(const char *buf) {
  guint32 event_length = 0;

  // event length is 4 bytes at position 9
  event_length = uint4korr(&buf[EVENT_LENGTH_POSITION]);
  // event length includes the header, plus a rotate event has a fixed 8byte
  // part we don't need
  event_length = event_length - EVENT_HEADER_LENGTH - EVENT_ROTATE_FIXED_LENGTH;

  return g_strndup(&buf[EVENT_HEADER_LENGTH + EVENT_ROTATE_FIXED_LENGTH],
                   event_length);
}

FILE *new_binlog_file(char *binlog_file, const char *binlog_dir) {
  FILE *outfile;
  char *filename;

  if (!compress_output) {
    filename = g_strdup_printf("%s/%s", binlog_dir, binlog_file);
    outfile = g_fopen(filename, "w");
  } else {
    filename = g_strdup_printf("%s/%s.gz", binlog_dir, binlog_file);
    outfile = (void *)gzopen(filename, "w");
  }
  g_free(filename);

  return outfile;
}

void close_binlog_file(FILE *outfile) {
  if (!compress_output)
    fclose(outfile);
  else
    gzclose((gzFile)outfile);
}

unsigned int get_event(const char *buf, unsigned int len) {
  if (len < EVENT_TYPE_POSITION)
    return EVENT_TOO_SHORT;
  return buf[EVENT_TYPE_POSITION];

  // TODO: Would be good if we can check for valid event type, unfortunately
  // this check can change from version to version
}

void write_binlog(FILE *file, const char *data, guint64 len) {
  int err;

  if (len > 0) {
    int write_result;

    if (!compress_output)
      write_result = write(fileno(file), data, len);
    else
      write_result = gzwrite((gzFile)file, data, len);

    if (write_result <= 0) {
      if (!compress_output)
        g_critical("Error: binlog: Error writing binary log: %s",
                   strerror(errno));
      else
        g_critical("Error: binlog: Error writing compressed binary log: %s",
                   gzerror((gzFile)file, &err));
    }
  }
}