Read data from collectl data files into a pandas DataFrame.
        :pattern: Absolute path to raw collectl files

def load_collectl(pattern, start_time, end_time):
    """Read data from collectl data files into a pandas DataFrame.
        :pattern: Absolute path to raw collectl files
    """
    start_tstamp = calendar.timegm(start_time.utctimetuple())
    end_tstamp = calendar.timegm(end_time.utctimetuple())

    cols = []
    rows = []

    for path in glob.glob(pattern):
        hardware, raw = _parse_raw(
            _CollectlGunzip(path, 'r'), start_tstamp, end_tstamp)

        if not cols:
            instances = {
                'disk': set(),
                'net': set(),
                'proc': set(),
            }
            for tstamp, sample in raw.items():
                for group, items in sample.items():
                    if group == 'disk':
                        instances['disk'] = instances['disk'].union(
                            items.keys())
                    elif group == 'net':
                        instances['net'] = instances['net'].union(
                            items.keys())
                    elif group == 'proc':
                        instances['proc'] = instances['proc'].union(
                            items.keys())

            cols = ['tstamp']
            cols.extend([
                'cpu_{}'.format(var)
                for var
                 in ['user', 'nice', 'sys', 'idle', 'wait',
                     'irq', 'soft', 'steal']
                 ])
            for node in instances['disk']:
                cols.extend([
                    '{}_{}'.format(node, var)
                    for var
                     in ['num_reads', 'reads_merged',
                         'sectors_read', 'msec_spent_reading',
                         'num_writes', 'writes_merged',
                         'sectors_written', 'msec_spent_writing',
                         'iops_in_progress', 'msec_spent_on_iops',
                         'weighted_msec_spent_on_iops']
                     ])
            cols.extend([
                'mem_{}'.format(var)
                for var
                 in ['total', 'free', 'buffers', 'cached']
                 ])
            for iface in instances['net']:
                cols.extend([
                    '{}_{}'.format(iface, var)
                    for var
                     in ['rbyte', 'rpkt', 'rerr', 'rdrop',
                         'rfifo', 'rframe', 'rcomp', 'rmulti',
                         'tbyte', 'tpkt', 'terr', 'tdrop',
                         'tfifo', 'tcoll', 'tcarrier', 'tcomp']
                     ])
            for pid in instances['proc']:
                cols.extend([
                    '{}_{}'.format(pid, var)
                    for var
                     in ['name', 'read_bytes', 'write_bytes']
                     ])

        for tstamp, sample in raw.items():
            if ('cpu' not in sample or
                'disk' not in sample or
                'mem' not in sample):
                # Skip incomplete samples; there might be a truncated
                # sample on the end of the file.
                continue

            values = [tstamp]
            values.extend([
                sample['cpu']['user'], sample['cpu']['nice'],
                sample['cpu']['sys'], sample['cpu']['idle'],
                sample['cpu']['wait'], sample['cpu']['irq'],
                sample['cpu']['soft'], sample['cpu']['steal'],
            ])
            for node in instances['disk']:
                data = sample['disk'].get(node, {})
                values.extend([
                    data.get('num_reads', 0),
                    data.get('reads_merged', 0),
                    data.get('sectors_read', 0),
                    data.get('msec_spent_reading', 0),
                    data.get('num_writes', 0),
                    data.get('writes_merged', 0),
                    data.get('sectors_written', 0),
                    data.get('msec_spent_writing', 0),
                    data.get('iops_in_progress', 0),
                    data.get('msec_spent_on_iops', 0),
                    data.get('weighted_msec_spent_on_iops', 0),
                ])
            values.extend([
                sample['mem']['total'], sample['mem']['free'],
                sample['mem']['buffers'], sample['mem']['cached'],
            ])
            for iface in instances['net']:
                data = sample['net'].get(iface, {})
                values.extend([
                    data.get('rbyte', 0), data.get('rpkt', 0),
                    data.get('rerr', 0), data.get('rdrop', 0),
                    data.get('rfifo', 0), data.get('rframe', 0),
                    data.get('rcomp', 0), data.get('rmulti', 0),
                    data.get('tbyte', 0), data.get('tpkt', 0),
                    data.get('terr', 0), data.get('tdrop', 0),
                    data.get('tfifo', 0), data.get('tcoll', 0),
                    data.get('tcarrier', 0), data.get('tcomp', 0),
                ])
            if 'proc' in sample:
                for pid in instances['proc']:
                    data = sample['proc'].get(pid, {})
                    values.extend([
                        data.get('cmd', ''),
                        data.get('read_bytes', 0),
                        data.get('write_bytes', 0),
                    ])

            rows.append(values)

    if len(rows) == 0:
        return pd.DataFrame(columns=cols), {}

    df = pd.DataFrame(rows, columns=cols)
    df = df.convert_objects(convert_numeric=True)
    df['tstamp'] = df['tstamp'].astype('datetime64[s]')
    df.set_index('tstamp', inplace=True)
    df = df.tz_localize('UTC')

    return df, hardware