You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@knox.apache.org by sm...@apache.org on 2021/03/14 23:11:46 UTC
[knox] branch master updated: KNOX-2549 - Fetching audit events
from CM beyond previous query timestamp (#412)
This is an automated email from the ASF dual-hosted git repository.
smolnar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/knox.git
The following commit(s) were added to refs/heads/master by this push:
new d0d5417 KNOX-2549 - Fetching audit events from CM beyond previous query timestamp (#412)
d0d5417 is described below
commit d0d5417d22f359d5910d934c37175fa00213b694
Author: Sandor Molnar <sm...@cloudera.com>
AuthorDate: Mon Mar 15 00:11:40 2021 +0100
KNOX-2549 - Fetching audit events from CM beyond previous query timestamp (#412)
To mitigate the chance of losing relevant audit events from CM, Knox needs to look back a little bit more than the last query timestamp. To be able to do this without processing relevant events again and again a cache was introduced to have them in memory for a short time.
Additionally, I removed made the maximum audit event resulset size so that every audit events are fetched from CM within the given time interval.
---
gateway-discovery-cm/pom.xml | 4 +
.../ClouderaManagerServiceDiscoveryMessages.java | 9 ++
.../cm/monitor/PollingConfigurationAnalyzer.java | 61 +++++++----
.../monitor/PollingConfigurationAnalyzerTest.java | 115 +++++++++++++++------
4 files changed, 133 insertions(+), 56 deletions(-)
diff --git a/gateway-discovery-cm/pom.xml b/gateway-discovery-cm/pom.xml
index 858b655..0d04bb3 100644
--- a/gateway-discovery-cm/pom.xml
+++ b/gateway-discovery-cm/pom.xml
@@ -101,6 +101,10 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ </dependency>
</dependencies>
</project>
diff --git a/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/ClouderaManagerServiceDiscoveryMessages.java b/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/ClouderaManagerServiceDiscoveryMessages.java
index 7c6782d..3934f95 100644
--- a/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/ClouderaManagerServiceDiscoveryMessages.java
+++ b/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/ClouderaManagerServiceDiscoveryMessages.java
@@ -172,6 +172,15 @@ public interface ClouderaManagerServiceDiscoveryMessages {
String discoveryAddress,
String sinceTimestamp);
+ @Message(level = MessageLevel.DEBUG, text = "There is no any activation event found within the given time period")
+ void noActivationEventFound();
+
+ @Message(level = MessageLevel.DEBUG, text = "Activation event relevance: {0} = {1}")
+ void activationEventRelevance(String eventId, String relevance);
+
+ @Message(level = MessageLevel.DEBUG, text = "Activation event - {0} - has already been processed, skipping ...")
+ void activationEventAlreadyProcessed(String eventId);
+
@Message(level = MessageLevel.DEBUG, text = "Analyzing current {0} configuration for changes...")
void analyzingCurrentServiceConfiguration(String serviceName);
diff --git a/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzer.java b/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzer.java
index 8f2d4d0..0918897 100644
--- a/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzer.java
+++ b/gateway-discovery-cm/src/main/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzer.java
@@ -29,6 +29,9 @@ import com.cloudera.api.swagger.model.ApiEventQueryResult;
import com.cloudera.api.swagger.model.ApiRole;
import com.cloudera.api.swagger.model.ApiRoleList;
import com.cloudera.api.swagger.model.ApiServiceConfig;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+
import org.apache.knox.gateway.GatewayServer;
import org.apache.knox.gateway.i18n.messages.MessagesFactory;
import org.apache.knox.gateway.services.GatewayServices;
@@ -57,6 +60,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.apache.knox.gateway.topology.discovery.ClusterConfigurationMonitor.ConfigurationChangeListener;
@@ -98,8 +102,7 @@ public class PollingConfigurationAnalyzer implements Runnable {
private static final int DEFAULT_POLLING_INTERVAL = 60;
- private static final ClouderaManagerServiceDiscoveryMessages log =
- MessagesFactory.get(ClouderaManagerServiceDiscoveryMessages.class);
+ private static final ClouderaManagerServiceDiscoveryMessages log = MessagesFactory.get(ClouderaManagerServiceDiscoveryMessages.class);
// Fully-qualified cluster name delimiter
private static final String FQCN_DELIM = "::";
@@ -120,18 +123,19 @@ public class PollingConfigurationAnalyzer implements Runnable {
// Polling interval in seconds
private int interval;
+ private final Cache<String, Long> processedEvents;
+
// Cache of ClouderaManager API clients, keyed by discovery address
private final Map<String, DiscoveryApiClient> clients = new ConcurrentHashMap<>();
// Timestamp records of the most recent start event query per discovery address
- private Map<String, String> eventQueryTimestamps = new ConcurrentHashMap<>();
+ private Map<String, Instant> eventQueryTimestamps = new ConcurrentHashMap<>();
// The amount of time before "now" to will check for start events the first time
private long eventQueryDefaultTimestampOffset = DEFAULT_EVENT_QUERY_DEFAULT_TIMESTAMP_OFFSET;
private boolean isActive;
-
PollingConfigurationAnalyzer(final ClusterConfigurationCache configCache,
final AliasService aliasService,
final KeystoreService keystoreService,
@@ -149,6 +153,7 @@ public class PollingConfigurationAnalyzer implements Runnable {
this.keystoreService = keystoreService;
this.changeListener = changeListener;
this.interval = interval;
+ this.processedEvents = Caffeine.newBuilder().expireAfterAccess(interval * 3, TimeUnit.SECONDS).maximumSize(1000).build();
}
void setInterval(int interval) {
@@ -229,6 +234,11 @@ public class PollingConfigurationAnalyzer implements Runnable {
boolean configHasChanged = false;
for (StartEvent re : relevantEvents) {
+ if (processedEvents.getIfPresent(re.auditEvent.getId()) != null) {
+ log.activationEventAlreadyProcessed(re.auditEvent.getId());
+ continue;
+ }
+
String serviceType = re.getServiceType();
if (CM_SERVICE_TYPE.equals(serviceType)) {
@@ -271,6 +281,10 @@ public class PollingConfigurationAnalyzer implements Runnable {
break; // No need to continue checking once we've identified one reason to perform discovery again
}
}
+
+ // these events should not be processed again even if the next CM query result contains them
+ relevantEvents.forEach(re -> processedEvents.put(re.auditEvent.getId(), 1L));
+
return configHasChanged;
}
@@ -354,10 +368,10 @@ public class PollingConfigurationAnalyzer implements Runnable {
}
void setEventQueryTimestamp(final String address, final String cluster, final Instant timestamp) {
- eventQueryTimestamps.put((address + ":" + cluster), timestamp.toString());
+ eventQueryTimestamps.put((address + ":" + cluster), timestamp);
}
- private String getEventQueryTimestamp(final String address, final String cluster) {
+ private Instant getEventQueryTimestamp(final String address, final String cluster) {
return eventQueryTimestamps.get(address + ":" + cluster);
}
@@ -383,25 +397,31 @@ public class PollingConfigurationAnalyzer implements Runnable {
List<StartEvent> relevantEvents = new ArrayList<>();
// Get the last event query timestamp
- String lastTimestamp = getEventQueryTimestamp(address, clusterName);
+ Instant lastTimestamp = getEventQueryTimestamp(address, clusterName);
// If this is the first query, then define the last timestamp
if (lastTimestamp == null) {
- lastTimestamp = Instant.now().minus(eventQueryDefaultTimestampOffset, ChronoUnit.MILLIS).toString();
+ lastTimestamp = Instant.now().minus(eventQueryDefaultTimestampOffset, ChronoUnit.MILLIS);
}
- log.queryingConfigActivationEventsFromCluster(clusterName, address, lastTimestamp);
+ // Go back in time an '2 x interval' more to mitigate the chance of losing a relevant audit event
+ lastTimestamp = lastTimestamp.minus(interval * 2, ChronoUnit.SECONDS);
+
+ log.queryingConfigActivationEventsFromCluster(clusterName, address, lastTimestamp.toString());
// Record the new event query timestamp for this address/cluster
setEventQueryTimestamp(address, clusterName, Instant.now());
// Query the event log from CM for service/cluster start events
- List<ApiEvent> events = queryEvents(getApiClient(configCache.getDiscoveryConfig(address, clusterName)),
- clusterName,
- lastTimestamp);
- for (ApiEvent event : events) {
- if(isRelevantEvent(event)) {
- relevantEvents.add(new StartEvent(event));
+ final List<ApiEvent> events = queryEvents(getApiClient(configCache.getDiscoveryConfig(address, clusterName)), clusterName, lastTimestamp.toString());
+
+ if (events.isEmpty()) {
+ log.noActivationEventFound();
+ } else {
+ for (ApiEvent event : events) {
+ if(isRelevantEvent(event)) {
+ relevantEvents.add(new StartEvent(event));
+ }
}
}
@@ -436,16 +456,13 @@ public class PollingConfigurationAnalyzer implements Runnable {
List<ApiEvent> events = new ArrayList<>();
// Setup the query for events
- String timeFilter =
- (since != null) ? String.format(Locale.ROOT, EVENTS_QUERY_TIMESTAMP_FORMAT, since) : "";
+ String timeFilter = (since != null) ? String.format(Locale.ROOT, EVENTS_QUERY_TIMESTAMP_FORMAT, since) : "";
- String queryString = String.format(Locale.ROOT,
- EVENTS_QUERY_FORMAT,
- clusterName,
- timeFilter);
+ String queryString = String.format(Locale.ROOT, EVENTS_QUERY_FORMAT, clusterName, timeFilter);
try {
- ApiEventQueryResult eventsResult = (new EventsResourceApi(client)).readEvents(20, queryString, 0);
+ // giving 'null' as maximum result size results in fetching all events from CM within the given time interval
+ ApiEventQueryResult eventsResult = (new EventsResourceApi(client)).readEvents(null, queryString, 0);
events.addAll(eventsResult.getItems());
} catch (ApiException e) {
log.clouderaManagerEventsAPIError(e);
diff --git a/gateway-discovery-cm/src/test/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzerTest.java b/gateway-discovery-cm/src/test/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzerTest.java
index 6f0a826..7e74ee6 100644
--- a/gateway-discovery-cm/src/test/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzerTest.java
+++ b/gateway-discovery-cm/src/test/java/org/apache/knox/gateway/topology/discovery/cm/monitor/PollingConfigurationAnalyzerTest.java
@@ -21,6 +21,7 @@ import com.cloudera.api.swagger.model.ApiEvent;
import com.cloudera.api.swagger.model.ApiEventAttribute;
import com.cloudera.api.swagger.model.ApiEventCategory;
import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.RandomStringUtils;
import org.apache.knox.gateway.GatewayServer;
import org.apache.knox.gateway.services.GatewayServices;
import org.apache.knox.gateway.services.ServiceType;
@@ -80,7 +81,7 @@ public class PollingConfigurationAnalyzerTest {
revisionEventAttrs.add(createEventAttribute("SERVICE", HiveOnTezServiceModelGenerator.SERVICE));
revisionEventAttrs.add(createEventAttribute("REVISION", "215"));
revisionEventAttrs.add(createEventAttribute("EVENTCODE", "EV_REVISION_CREATED"));
- final ApiEvent revisionEvent = createApiEvent(ApiEventCategory.AUDIT_EVENT, revisionEventAttrs);
+ final ApiEvent revisionEvent = createApiEvent(ApiEventCategory.AUDIT_EVENT, revisionEventAttrs, null);
doTestEventWithoutConfigChange(revisionEvent, clusterName);
}
@@ -247,6 +248,29 @@ public class PollingConfigurationAnalyzerTest {
}
@Test
+ public void testRestartWaitingForStalenessSuccessEventProcessedOnlyOnce() {
+ final String address = "http://host1:1234";
+ final String clusterName = "Cluster 9";
+
+ // Simulate a successful restart waiting for staleness event with id = 123
+ final ApiEvent rollingRestartEvent = createApiEvent(clusterName, PollingConfigurationAnalyzer.CM_SERVICE_TYPE, PollingConfigurationAnalyzer.CM_SERVICE,
+ PollingConfigurationAnalyzer.RESTART_WAITING_FOR_STALENESS_SUCCESS_COMMAND, PollingConfigurationAnalyzer.SUCCEEDED_STATUS, "EV_CLUSTER_RESTARTED",
+ "123");
+
+ final ChangeListener listener = new ChangeListener();
+ final TestablePollingConfigAnalyzer pca = buildPollingConfigAnalyzer(address, clusterName, Collections.emptyMap(), listener);
+
+ // this should trigger a change notification
+ doTestEvent(rollingRestartEvent, address, clusterName, Collections.emptyMap(), Collections.emptyMap(), pca);
+ assertTrue("Expected a change notification", listener.wasNotified(address, clusterName));
+
+ // this should NOT trigger a notification as the same event has already been processed
+ listener.clearNotification();
+ doTestEvent(rollingRestartEvent, address, clusterName, Collections.emptyMap(), Collections.emptyMap(), pca);
+ assertFalse("Unexpected change notification", listener.wasNotified(address, clusterName));
+ }
+
+ @Test
public void testClusterConfigMonitorTerminationForNoLongerReferencedClusters() {
final String address = "http://host1:1234";
final String clusterName = "Cluster 7";
@@ -360,7 +384,7 @@ public class PollingConfigurationAnalyzerTest {
apiEventAttrs.add(createEventAttribute("CLUSTER", clusterName));
apiEventAttrs.add(createEventAttribute("SERVICE_TYPE", serviceType));
apiEventAttrs.add(createEventAttribute("SERVICE", service));
- ApiEvent apiEvent = createApiEvent(category, apiEventAttrs);
+ ApiEvent apiEvent = createApiEvent(category, apiEventAttrs, null);
PollingConfigurationAnalyzer.StartEvent restartEvent = new PollingConfigurationAnalyzer.StartEvent(apiEvent);
assertNotNull(restartEvent);
@@ -370,34 +394,18 @@ public class PollingConfigurationAnalyzerTest {
assertNotNull(restartEvent.getTimestamp());
}
- private ChangeListener doTestEvent(final ApiEvent event,
- final String address,
- final String clusterName,
- final Map<String, ServiceConfigurationModel> serviceConfigurationModels,
- final Map<String, ServiceConfigurationModel> updatedServiceConfigurationModels) {
- // Mock the service discovery details
- ServiceDiscoveryConfig sdc = EasyMock.createNiceMock(ServiceDiscoveryConfig.class);
- EasyMock.expect(sdc.getCluster()).andReturn(clusterName).anyTimes();
- EasyMock.expect(sdc.getAddress()).andReturn(address).anyTimes();
- EasyMock.expect(sdc.getUser()).andReturn("u").anyTimes();
- EasyMock.expect(sdc.getPasswordAlias()).andReturn("a").anyTimes();
- EasyMock.replay(sdc);
-
- final Map<String, List<String>> clusterNames = new HashMap<>();
- clusterNames.put(address, Collections.singletonList(clusterName));
+ private ChangeListener doTestEvent(final ApiEvent event, final String address, final String clusterName,
+ final Map<String, ServiceConfigurationModel> serviceConfigurationModels, final Map<String, ServiceConfigurationModel> updatedServiceConfigurationModels) {
+ return doTestEvent(event, address, clusterName, serviceConfigurationModels, updatedServiceConfigurationModels, null);
+ }
- // Mock a ClusterConfigurationCache for the monitor to use
- ClusterConfigurationCache configCache = EasyMock.createNiceMock(ClusterConfigurationCache.class);
- EasyMock.expect(configCache.getDiscoveryConfig(address, clusterName)).andReturn(sdc).anyTimes();
- EasyMock.expect(configCache.getClusterNames()).andReturn(clusterNames).anyTimes();
- EasyMock.expect(configCache.getClusterServiceConfigurations(address, clusterName))
- .andReturn(serviceConfigurationModels)
- .anyTimes();
- EasyMock.replay(configCache);
+ private ChangeListener doTestEvent(final ApiEvent event, final String address, final String clusterName,
+ final Map<String, ServiceConfigurationModel> serviceConfigurationModels, final Map<String, ServiceConfigurationModel> updatedServiceConfigurationModels,
+ final TestablePollingConfigAnalyzer pollingConfigAnalyzer) {
// Create the monitor, registering a listener so we can verify that change notification works
- ChangeListener listener = new ChangeListener();
- TestablePollingConfigAnalyzer pca = new TestablePollingConfigAnalyzer(configCache, listener);
+ final ChangeListener listener = new ChangeListener();
+ final TestablePollingConfigAnalyzer pca = pollingConfigAnalyzer == null ? buildPollingConfigAnalyzer(address, clusterName, serviceConfigurationModels, listener) : pollingConfigAnalyzer;
pca.setInterval(5);
// Add updated service config models
@@ -424,6 +432,29 @@ public class PollingConfigurationAnalyzerTest {
return listener;
}
+ private TestablePollingConfigAnalyzer buildPollingConfigAnalyzer(final String address, final String clusterName,
+ final Map<String, ServiceConfigurationModel> serviceConfigurationModels, ChangeListener listener) {
+ // Mock the service discovery details
+ ServiceDiscoveryConfig sdc = EasyMock.createNiceMock(ServiceDiscoveryConfig.class);
+ EasyMock.expect(sdc.getCluster()).andReturn(clusterName).anyTimes();
+ EasyMock.expect(sdc.getAddress()).andReturn(address).anyTimes();
+ EasyMock.expect(sdc.getUser()).andReturn("u").anyTimes();
+ EasyMock.expect(sdc.getPasswordAlias()).andReturn("a").anyTimes();
+ EasyMock.replay(sdc);
+
+ final Map<String, List<String>> clusterNames = new HashMap<>();
+ clusterNames.put(address, Collections.singletonList(clusterName));
+
+ // Mock a ClusterConfigurationCache for the monitor to use
+ ClusterConfigurationCache configCache = EasyMock.createNiceMock(ClusterConfigurationCache.class);
+ EasyMock.expect(configCache.getDiscoveryConfig(address, clusterName)).andReturn(sdc).anyTimes();
+ EasyMock.expect(configCache.getClusterNames()).andReturn(clusterNames).anyTimes();
+ EasyMock.expect(configCache.getClusterServiceConfigurations(address, clusterName)).andReturn(serviceConfigurationModels).anyTimes();
+ EasyMock.replay(configCache);
+
+ return new TestablePollingConfigAnalyzer(configCache, listener);
+ }
+
private void doTestEventWithConfigChange(final ApiEvent event, final String clusterName) {
final String address = "http://host1:1234";
@@ -475,8 +506,7 @@ public class PollingConfigurationAnalyzerTest {
nnRoleProps.put(nsPropertyName, nsValue);
nnRoleProps.put(portPropertyName, portValue);
nnRoleConf.put(NameNodeServiceModelGenerator.ROLE_TYPE, nnRoleProps);
- serviceConfigurationModels.put(NameNodeServiceModelGenerator.SERVICE_TYPE,
- createModel(nnServiceConf, nnRoleConf));
+ serviceConfigurationModels.put(NameNodeServiceModelGenerator.SERVICE_TYPE, createModel(nnServiceConf, nnRoleConf));
// Create another version of the same ServiceConfigurationModel with unmodified property values
ServiceConfigurationModel updatedNNModel = new ServiceConfigurationModel();
@@ -510,16 +540,24 @@ public class PollingConfigurationAnalyzerTest {
final String service,
final String command,
final String commandStatues) {
-
return createApiEvent(clusterName, serviceType, service, command, commandStatues, "");
}
private ApiEvent createApiEvent(final String clusterName,
+ final String serviceType,
+ final String service,
+ final String command,
+ final String commandStatues,
+ final String eventCode) {
+ return createApiEvent(clusterName, serviceType, service, command, commandStatues, eventCode, null);
+ }
+ private ApiEvent createApiEvent(final String clusterName,
final String serviceType,
final String service,
final String command,
final String commandStatues,
- final String eventCode) {
+ final String eventCode,
+ final String id) {
List<ApiEventAttribute> attrs = new ArrayList<>();
attrs.add(createEventAttribute("CLUSTER", clusterName));
attrs.add(createEventAttribute("SERVICE_TYPE", serviceType));
@@ -527,11 +565,16 @@ public class PollingConfigurationAnalyzerTest {
attrs.add(createEventAttribute("COMMAND", command));
attrs.add(createEventAttribute("COMMAND_STATUS", commandStatues));
attrs.add(createEventAttribute("EVENTCODE", eventCode));
- return createApiEvent(ApiEventCategory.AUDIT_EVENT, attrs);
+ return createApiEvent(ApiEventCategory.AUDIT_EVENT, attrs, id);
}
- private ApiEvent createApiEvent(final ApiEventCategory category, final List<ApiEventAttribute> attrs) {
+ private ApiEvent createApiEvent(final ApiEventCategory category, final List<ApiEventAttribute> attrs, String id) {
ApiEvent event = EasyMock.createNiceMock(ApiEvent.class);
+ if (id == null) {
+ EasyMock.expect(event.getId()).andReturn(RandomStringUtils.random(8, true, true)).anyTimes();
+ } else {
+ EasyMock.expect(event.getId()).andReturn(id).anyTimes();
+ }
EasyMock.expect(event.getTimeOccurred()).andReturn(Instant.now().toString()).anyTimes();
EasyMock.expect(event.getCategory()).andReturn(category).anyTimes();
EasyMock.expect(event.getAttributes()).andReturn(attrs).anyTimes();
@@ -580,7 +623,7 @@ public class PollingConfigurationAnalyzerTest {
TestablePollingConfigAnalyzer(ClusterConfigurationCache cache,
ConfigurationChangeListener listener) {
- super(cache, null, null, listener);
+ super(cache, null, null, listener, 20);
}
void addRestartEvent(final String service, final ApiEvent restartEvent) {
@@ -623,6 +666,10 @@ public class PollingConfigurationAnalyzerTest {
return clusterName.equals(notifications.get(source));
}
+ void clearNotification() {
+ notifications.clear();
+ }
+
int howManyNotifications(final String source, final String clusterName) {
return events.size();
}