| 1443 | |
| 1444 | |
| 1445 | void Master::consume(MessageEvent&& event) |
| 1446 | { |
| 1447 | // There are three cases about the message's UPID with respect to |
| 1448 | // 'frameworks.principals': |
| 1449 | // 1) if a <UPID, principal> pair exists and the principal is Some, |
| 1450 | // it's a framework with its principal specified. |
| 1451 | // 2) if a <UPID, principal> pair exists and the principal is None, |
| 1452 | // it's a framework without a principal. |
| 1453 | // 3) if a <UPID, principal> pair does not exist in the map, it's |
| 1454 | // either an unregistered framework or not a framework. |
| 1455 | // The logic for framework message counters and rate limiting |
| 1456 | // mainly concerns with whether the UPID is a *registered* |
| 1457 | // framework and whether the framework has a principal so we use |
| 1458 | // these two temp variables to simplify the condition checks below. |
| 1459 | bool isRegisteredFramework = |
| 1460 | frameworks.principals.contains(event.message.from); |
| 1461 | const Option<string> principal = isRegisteredFramework |
| 1462 | ? frameworks.principals[event.message.from] |
| 1463 | : Option<string>::none(); |
| 1464 | |
| 1465 | // Increment the "message_received" counter if the message is from |
| 1466 | // a framework and such a counter is configured for it. |
| 1467 | // See comments for 'Master::Metrics::Frameworks' and |
| 1468 | // 'Master::Frameworks::principals' for details. |
| 1469 | if (principal.isSome()) { |
| 1470 | // If the framework has a principal, the counter must exist. |
| 1471 | CHECK(metrics->frameworks.contains(principal.get())); |
| 1472 | Counter messages_received = |
| 1473 | metrics->frameworks.at(principal.get())->messages_received; |
| 1474 | ++messages_received; |
| 1475 | } |
| 1476 | |
| 1477 | // All messages are filtered when non-leading. |
| 1478 | if (!elected()) { |
| 1479 | VLOG(1) << "Dropping '" << event.message.name << "' message since " |
| 1480 | << "not elected yet"; |
| 1481 | |
| 1482 | ++metrics->dropped_messages; |
| 1483 | return; |
| 1484 | } |
| 1485 | |
| 1486 | CHECK_SOME(recovered); |
| 1487 | |
| 1488 | // All messages are filtered while recovering. |
| 1489 | // TODO(bmahler): Consider instead re-enqueing *all* messages |
| 1490 | // through recover(). What are the performance implications of |
| 1491 | // the additional queueing delay and the accumulated backlog |
| 1492 | // of messages post-recovery? |
| 1493 | if (!recovered->isReady()) { |
| 1494 | VLOG(1) << "Dropping '" << event.message.name << "' message since " |
| 1495 | << "not recovered yet"; |
| 1496 | |
| 1497 | ++metrics->dropped_messages; |
| 1498 | return; |
| 1499 | } |
| 1500 | |
| 1501 | // Throttle the message if it's a framework message and a |
| 1502 | // RateLimiter is configured for the framework's principal. |