为 appengine 设计访问/网络统计计数器模块

Question

我需要一个 appengine 的访问统计模块来跟踪一些请求处理程序并将统计信息收集到 bigtable。我没有在 github 上找到任何现成的解决方案，Google 的示例要么过于简单（memcached frontpage counter with cron），要么过度杀伤（accurate sharded counter）。但最重要的是，其他地方讨论的 appengine-counter 解决方案都没有包含统计所需的时间组件（每小时、每天计数）。

要求：系统不需要 100% 准确，可以忽略 memcache 丢失（如果不常见）。这应该大大简化事情。这个想法是只使用内存缓存并按时间间隔累积统计信息。

用例：您系统上的用户创建内容（例如页面）。你想跟踪大约。 每小时或每天查看用户页面的频率。有些页面经常查看，有些则从不查看。您想按用户和时间范围查询。子页面可能有固定的 ID（查询主页上点击次数最多的用户）。您可能想要删除旧条目（查询年份=xxxx 的条目）。

class StatisticsDB(ndb.Model):
    # key.id() = something like YYYY-MM-DD-HH_groupId_countableID ... contains date
    # timeframeId = ndb.StringProperty() YYYY-MM-DD-HH needed for cleanup if counter uses ancestors
    countableId = ndb.StringProperty(required=True) # name of counter within group
    groupId = ndb.StringProperty() # counter group (allows single DB query with timeframe prefix inequality)
    count = ndb.Integerproperty() # count per specified timeframe

    @classmethod
    def increment(class, groupID, countableID):
        # increment memcache
        # save hourly to DB (see below)

注意：groupId 和 countableId 索引对于避免查询中的 2 个不等式是必要的。 （查询一个 groupId/userId 和 chart/highcount-query 的所有可数：countableId 具有最高计数派生 groupId/user），在 DB 中使用祖先可能不支持图表查询。

问题是如何最好地将memcached计数器保存到DB：

1. cron：示例文档 (example front-page counter) 中提到了这种方法，但使用在 cron 处理程序中硬编码的固定计数器 ID。由于没有对现有 memcache 键的前缀查询，因此确定在最后一个时间间隔内在 memcache 中创建了哪些计数器 ID 并且需要保存可能是瓶颈。
2. task-queue：如果创建了一个计数器，则安排一个任务来收集它并将其写入 DB。 COST：队列处理程序保存数据时，每个使用的计数器有 1 个任务队列条目，每个时间粒度（例如 1 小时）有一个 ndb.put。似乎也是最有希望准确捕捉偶发事件的方法。
3. increment(id) 执行时不经常：如果新的时间范围开始，则保存前一个时间范围。这需要每次增量至少 2 次 memcache 访问（获取日期、增量计数器）。一个用于跟踪时间范围，一个用于计数器。缺点：过期时间较长的突发计数器可能会丢失缓存。
4. increment(id) 执行时不经常：概率性：如果随机 % 100 == 0 则保存到 DB，但计数器应具有均匀分布的计数事件
5. increment(id) 执行时不经常：如果计数器达到例如100 然后保存到数据库

有没有人解决这个问题，设计这个的好方法是什么？ 每种方法的弱点和优势是什么？ 这里是否缺少替代方法？

假设：计数可能会稍微不准确（缓存丢失），counterID 空间很大，counterID 偶尔递增（有些每天一次，有些经常每天）

更新：1）我认为可以使用类似于任务队列的cron。只需使用 memcached=True 创建计数器的数据库模型，并在 cron 中为所有以这种方式标记的计数器运行查询。成本：1 次增量，cron 查询，1 次更新计数器。如果没有充分考虑，这似乎比任务方法更昂贵/复杂。

在别处讨论过：

• High concurrency non-sharded counters - 每个时间段不计数
• Open Source GAE Fast Counters - 每个时间范围内没有计数，与分片解决方案的性能相比很好，报告了由于 memcache 丢失而导致的预期损失

Answer 1

是的，您的第二个想法似乎最能满足您的要求。

要实现它，您需要一个指定延迟的任务执行。

我为此使用了the deferred library，使用deferred.defer() 的countdown 参数。与此同时，我了解到标准队列库具有类似的支持，通过为 Task 构造函数指定 countdown 参数（我还没有使用这种方法）。

因此，每当您创建内存缓存计数器时，也会将延迟执行任务排入队列（将计数器的内存缓存键传入其有效负载），这将：

• 使用任务负载中的键获取内存缓存计数器值
• 将值添加到相应的数据库计数器
• 数据库更新成功后删除memcache计数器

在任务执行中读取内存缓存计数器和删除内存缓存计数器之间，您可能会丢失并发请求的增量。您可以通过在读取内存缓存计数器后立即删除它来减少此类损失，但如果数据库更新由于任何原因失败，您将面临丢失整个计数的风险 - 重新尝试任务将不再找到内存缓存计数器。如果这些都不令人满意，您可以进一步完善解决方案：

延迟的任务：

• 读取内存缓存计数器值
• 将另一个（事务性）任务（无延迟）排入队列，以将值添加到数据库计数器
• 删除内存缓存计数器

非延迟任务现在是幂等的，可以安全地重试直到成功。

并发请求丢失增量的风险仍然存在，但我想它更小。

更新：

Task Queues 优于延迟库，延迟功能可通过 taskqueue.add() 的可选 countdown 或 eta 参数获得：

• 倒计时 -- 该任务应该运行或租用的未来时间（以秒为单位）。默认为零。如果不指定此参数 你指定了一个 eta。

• eta -- 一个datetime.datetime，它指定任务应该运行的绝对最早时间。您不能指定此参数，如果 指定倒计时参数。这个论点可以是时间 zone-aware 或 time zone-naive，或设置为过去的时间。如果 参数设置为无，默认值为现在。对于拉任务，没有 工人可以在 eta 指示的时间之前租用任务 论据。

Answer 2

在分布式系统中计算事物是一个难题。在 App Engine 的早期，有一些关于这个问题的好信息。我会从 Sharding Counter 开始，尽管它是在 2008 年编写的，但仍然具有相关性。

Answer 3

这里是实现每小时时间框架的任务队列方法的代码。有趣的是，它在没有事务和其他互斥魔法的情况下也能工作。 （为了便于阅读，python 方法的缩进是错误的。）

支持priorities for memcache 将提高此解决方案的准确性。

TASK_URL = '/h/statistics/collect/' # Example: '/h/statistics/collect/{counter-id}"?groupId=" + groupId + "&countableId=" + countableId'
MEMCACHE_PREFIX = "StatisticsDB_"

class StatisticsDB(ndb.Model):
"""
Memcached counting saved each hour to DB.
"""
    # key.id() = 2016-01-31-17_groupId_countableId
    countableId = ndb.StringProperty(required=True) # unique name of counter within group
    groupId = ndb.StringProperty() # couter group (allows single DB query for group of counters)
    count = ndb.IntegerProperty(default=0) # count per timeframe


@classmethod
def increment(cls, groupId, countableId):  # throws InvalidTaskNameError
    """
    Increment a counter. countableId is the unique id of the countable
    throws InvalidTaskNameError if ids do not match: [a-zA-Z0-9-_]{1,500}
    """
    # Calculate memcache key and db_key at this time
    # the counting timeframe is 1h, determined by %H, MUST MATCH ETA calculation in _add_task()
    counter_key = datetime.datetime.utcnow().strftime("%Y-%m-%d-%H") + "_" + groupId +"_"+ countableId;
    client = memcache.Client()

    n = client.incr(MEMCACHE_PREFIX + counter_key)
    if n is None:
        cls._add_task(counter_key, groupId, countableId)
        client.incr(MEMCACHE_PREFIX + counter_key, initial_value=0)


@classmethod
def _add_task(cls, counter_key, groupId, countableId):
    taskurl = TASK_URL + counter_key + "?groupId=" + groupId + "&countableId=" + countableId
    now = datetime.datetime.now()
    # the counting timeframe is 1h, determined by counter_key, MUST MATCH ETA calculation
    eta = now + datetime.timedelta(minutes = (61-now.minute)) # at most 1h later, randomized over 1 minute, throttled by queue parameters
    task = taskqueue.Task(url=taskurl, method='GET', name=MEMCACHE_PREFIX + counter_key, eta=eta)
    queue = taskqueue.Queue(name='StatisticsDB')
    try:
        queue.add(task)
    except taskqueue.TaskAlreadyExistsError: # may also occur if 2 increments are done simultaneously
        logging.warning("StatisticsDB TaskAlreadyExistsError lost memcache for %s", counter_key)
    except taskqueue.TombstonedTaskError: # task name is locked for ...
        logging.warning("StatisticsDB TombstonedTaskError some bad guy ran this task premature manually %s", counter_key)


@classmethod
def save2db_task_handler(cls, counter_key, countableId, groupId):
    """
    Save counter from memcache to DB. Idempotent method.
    At the time this executes no more increments to this counter occur.
    """
    dbkey = ndb.Key(StatisticsDB, counter_key)

    n = memcache.get(MEMCACHE_PREFIX + counter_key)        
    if n is None:
        logging.warning("StatisticsDB lost count for %s", counter_key)
        return

    stats = StatisticsDB(key=dbkey, count=n, countableId=countableId, groupId=groupId)
    stats.put()
    memcache.delete(MEMCACHE_PREFIX + counter_key) # delete if put succeeded
    logging.info("StatisticsDB saved %s n = %i", counter_key, n)