export a given log group to s3

def export(group, bucket, prefix, start, end, role, poll_period=120,
           session=None, name="", region=None):
    """export a given log group to s3"""
    start = start and isinstance(start, six.string_types) and parse(start) or start
    end = (end and isinstance(start, six.string_types) and
           parse(end) or end or datetime.now())
    start = start.replace(tzinfo=tzlocal()).astimezone(tzutc())
    end = end.replace(tzinfo=tzlocal()).astimezone(tzutc())

    if session is None:
        session = get_session(role, region)

    client = session.client('logs')

    paginator = client.get_paginator('describe_log_groups')
    for p in paginator.paginate():
        found = False
        for _group in p['logGroups']:
            if _group['logGroupName'] == group:
                group = _group
                found = True
                break
        if found:
            break

    if not found:
        raise ValueError("Log group %s not found." % group)

    if prefix:
        prefix = "%s/%s" % (prefix.rstrip('/'), group['logGroupName'].strip('/'))
    else:
        prefix = group['logGroupName']

    named_group = "%s:%s" % (name, group['logGroupName'])
    log.info(
        "Log exporting group:%s start:%s end:%s bucket:%s prefix:%s size:%s",
        named_group,
        start.strftime('%Y/%m/%d'),
        end.strftime('%Y/%m/%d'),
        bucket,
        prefix,
        group['storedBytes'])

    t = time.time()
    days = [(
        start + timedelta(i)).replace(minute=0, hour=0, second=0, microsecond=0)
        for i in range((end - start).days)]
    day_count = len(days)
    s3 = boto3.Session().client('s3')
    days = filter_extant_exports(s3, bucket, prefix, days, start, end)

    log.info("Group:%s filtering s3 extant keys from %d to %d start:%s end:%s",
             named_group, day_count, len(days),
             days[0] if days else '', days[-1] if days else '')
    t = time.time()

    retry = get_retry(('SlowDown',))

    for idx, d in enumerate(days):
        date = d.replace(minute=0, microsecond=0, hour=0)
        export_prefix = "%s%s" % (prefix, date.strftime("/%Y/%m/%d"))
        params = {
            'taskName': "%s-%s" % ("c7n-log-exporter",
                                   date.strftime("%Y-%m-%d")),
            'logGroupName': group['logGroupName'],
            'fromTime': int(time.mktime(
                date.replace(
                    minute=0, microsecond=0, hour=0).timetuple()) * 1000),
            'to': int(time.mktime(
                date.replace(
                    minute=59, hour=23, microsecond=0).timetuple()) * 1000),
            'destination': bucket,
            'destinationPrefix': export_prefix
        }

        # if stream_prefix:
        #    params['logStreamPrefix'] = stream_prefix
        try:
            s3.head_object(Bucket=bucket, Key=prefix)
        except ClientError as e:
            if e.response['Error']['Code'] != '404':  # Not Found
                raise
            s3.put_object(
                Bucket=bucket,
                Key=prefix,
                Body=json.dumps({}),
                ACL="bucket-owner-full-control",
                ServerSideEncryption="AES256")

        t = time.time()
        counter = 0
        while True:
            counter += 1
            try:
                result = client.create_export_task(**params)
            except ClientError as e:
                if e.response['Error']['Code'] == 'LimitExceededException':
                    time.sleep(poll_period)
                    # log every 30m of export waiting
                    if counter % 6 == 0:
                        log.debug(
                            "group:%s day:%s waiting for %0.2f minutes",
                            named_group, d.strftime('%Y-%m-%d'),
                            (counter * poll_period) / 60.0)
                    continue
                raise
            retry(
                s3.put_object_tagging,
                Bucket=bucket, Key=prefix,
                Tagging={
                    'TagSet': [{
                        'Key': 'LastExport',
                        'Value': d.isoformat()}]})
            break

        log.info(
            "Log export time:%0.2f group:%s day:%s bucket:%s prefix:%s task:%s",
            time.time() - t,
            named_group,
            d.strftime("%Y-%m-%d"),
            bucket,
            params['destinationPrefix'],
            result['taskId'])

    log.info(
        ("Exported log group:%s time:%0.2f days:%d start:%s"
         " end:%s bucket:%s prefix:%s"),
        named_group,
        time.time() - t,
        len(days),
        start.strftime('%Y/%m/%d'),
        end.strftime('%Y/%m/%d'),
        bucket,
        prefix)