diff --git a/hawkbit-autoconfigure/src/main/java/org/eclipse/hawkbit/autoconfigure/repository/event/EventPublisherAutoConfiguration.java b/hawkbit-autoconfigure/src/main/java/org/eclipse/hawkbit/autoconfigure/repository/event/EventPublisherAutoConfiguration.java index 6a4e98242..64fe11c06 100644 --- a/hawkbit-autoconfigure/src/main/java/org/eclipse/hawkbit/autoconfigure/repository/event/EventPublisherAutoConfiguration.java +++ b/hawkbit-autoconfigure/src/main/java/org/eclipse/hawkbit/autoconfigure/repository/event/EventPublisherAutoConfiguration.java @@ -15,7 +15,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; /** - * Autoconfiguration for the event bus. + * Autoconfiguration for the events.. */ @Configuration @Import(EventPublisherConfiguration.class) diff --git a/hawkbit-ddi/hawkbit-ddi-resource/src/test/java/org/eclipse/hawkbit/ddi/rest/resource/DdiConfirmationBaseTest.java b/hawkbit-ddi/hawkbit-ddi-resource/src/test/java/org/eclipse/hawkbit/ddi/rest/resource/DdiConfirmationBaseTest.java index d2ce3f180..3d6b54b21 100644 --- a/hawkbit-ddi/hawkbit-ddi-resource/src/test/java/org/eclipse/hawkbit/ddi/rest/resource/DdiConfirmationBaseTest.java +++ b/hawkbit-ddi/hawkbit-ddi-resource/src/test/java/org/eclipse/hawkbit/ddi/rest/resource/DdiConfirmationBaseTest.java @@ -262,7 +262,7 @@ class DdiConfirmationBaseTest extends AbstractDDiApiIntegrationTest { @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 3), // implicit lock @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionCreatedEvent.class, count = 1), + @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = ActionUpdatedEvent.class, count = 2), @Expect(type = TargetUpdatedEvent.class, count = 1), @Expect(type = TargetUpdatedEvent.class, count = 1), @@ -440,7 +440,7 @@ class DdiConfirmationBaseTest extends AbstractDDiApiIntegrationTest { @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 3), // implicit lock @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionCreatedEvent.class, count = 1), + @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = ActionUpdatedEvent.class, count = 2), @Expect(type = TargetUpdatedEvent.class, count = 1), @Expect(type = TenantConfigurationCreatedEvent.class, count = 1) }) diff --git a/hawkbit-ddi/hawkbit-ddi-resource/src/test/resources/ddi-test.properties b/hawkbit-ddi/hawkbit-ddi-resource/src/test/resources/ddi-test.properties index 2db581c1c..06903de8c 100644 --- a/hawkbit-ddi/hawkbit-ddi-resource/src/test/resources/ddi-test.properties +++ b/hawkbit-ddi/hawkbit-ddi-resource/src/test/resources/ddi-test.properties @@ -25,6 +25,4 @@ spring.servlet.multipart.max-file-size=5MB hawkbit.server.security.dos.maxStatusEntriesPerAction=100 hawkbit.server.security.dos.maxAttributeEntriesPerTarget=10 # Quota - END - -# disable spring cloud bus for tests -spring.cloud.bus.enabled=false +org.eclipse.hawkbit.events.remote-enabled=false diff --git a/hawkbit-ddi/hawkbit-ddi-server/README.md b/hawkbit-ddi/hawkbit-ddi-server/README.md index 0d230c15a..663f5168e 100644 --- a/hawkbit-ddi/hawkbit-ddi-server/README.md +++ b/hawkbit-ddi/hawkbit-ddi-server/README.md @@ -24,37 +24,6 @@ run org.eclipse.hawkbit.app.ddi.DDIStart The Management API can be accessed via http://localhost:8081/rest/v1 The root url http://localhost:8081 will redirect directly to the Swagger Management UI -### Clustering (Experimental!!!) - -The micro-service instances are configured to communicate via Spring Cloud Bus. You could run multiple instances of any -micro-service but hawkbit-mgmt-server. Management server run some schedulers which shall not run simultaneously - e.g. -auto assignment checker and rollouts executor. To run multiple management server instances you shall do some extensions -of hawkbit to ensure that they wont run schedulers simultaneously or you shall configure all instances but one to do not -run schedulers! - -## Optional Protostuff for Spring cloud bus - -The micro-service instances are configured to communicate via Spring Cloud Bus. Optionally, you could -use [Protostuff](https://github.com/protostuff/protostuff) based message payload serialization for improved performance. - -**Note**: If Protostuff is enabled it shall be enabled on all microservices! - -Add/Uncomment to/in your `application.properties` : - -```properties -spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff -``` - -Add to your `pom.xml` : - -```xml - - io.protostuff - protostuff-core - - - io.protostuff - protostuff-runtime - -``` \ No newline at end of file +# Clustering (Experimental!!!) +## Remote Events between micro-services +[See more information](../../site/content/guides/clustering.md) \ No newline at end of file diff --git a/hawkbit-ddi/hawkbit-ddi-server/src/main/resources/application.properties b/hawkbit-ddi/hawkbit-ddi-server/src/main/resources/application.properties index a26f83e3f..6c949f01c 100644 --- a/hawkbit-ddi/hawkbit-ddi-server/src/main/resources/application.properties +++ b/hawkbit-ddi/hawkbit-ddi-server/src/main/resources/application.properties @@ -14,7 +14,6 @@ spring.main.allow-bean-definition-overriding=true server.port=8081 # Logging configuration -logging.level.org.eclipse.hawkbit.eventbus.DeadEventListener=WARN logging.level.org.springframework.boot.actuate.audit.listener.AuditListener=WARN logging.level.org.hibernate.validator.internal.util.Version=WARN # security Log with hints on potential attacks @@ -50,15 +49,10 @@ hawkbit.lock=inMemory # Disable discovery client of spring-cloud-commons spring.cloud.discovery.enabled=false -# Configure communication between services -endpoints.spring.cloud.bus.refresh.enabled=false -endpoints.spring.cloud.bus.env.enabled=false -spring.cloud.stream.bindings.springCloudBusInput.group=ddi-server - -# To use protostuff (for instance fot improved performance) you shall uncomment -# the following two lines and add io.protostuff:protostuff-core and io.protostuff:protostuff-runtime to dependencies -#spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -#spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff +# remote events configuration +spring.config.import=classpath:/hawkbit-events-defaults.properties +# Optional: Use protostuff (if enabled) +# spring.cloud.stream.default.content-type=application/binary+protostuff # Swagger Configuration / https://springdoc.org/v2/#properties springdoc.api-docs.version=openapi_3_0 diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java index f5d2a80c2..b47f3df6a 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java +++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java @@ -55,7 +55,14 @@ import org.eclipse.hawkbit.repository.SystemManagement; import org.eclipse.hawkbit.repository.TargetManagement; import org.eclipse.hawkbit.repository.TenantConfigurationManagement; import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent; -import org.eclipse.hawkbit.repository.event.remote.MultiActionEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionCancelServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent; import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent; import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent; import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent; @@ -73,8 +80,6 @@ import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageBuilder; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; -import org.springframework.cloud.bus.ServiceMatcher; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; import org.springframework.context.event.EventListener; import org.springframework.data.domain.PageRequest; import org.springframework.security.core.context.SecurityContext; @@ -96,7 +101,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { private final SystemSecurityContext systemSecurityContext; private final SystemManagement systemManagement; private final TargetManagement targetManagement; - private final ServiceMatcher serviceMatcher; private final SoftwareModuleManagement softwareModuleManagement; private final DistributionSetManagement distributionSetManagement; private final DeploymentManagement deploymentManagement; @@ -111,7 +115,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { * @param systemSecurityContext for execution with system permissions * @param systemManagement the systemManagement * @param targetManagement to access target information - * @param serviceMatcher to check in cluster case if the message is from the same cluster node * @param distributionSetManagement to retrieve modules * @param tenantConfigurationManagement to access tenant configuration */ @@ -120,7 +123,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { final RabbitTemplate rabbitTemplate, final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler, final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement, - final TargetManagement targetManagement, final ServiceMatcher serviceMatcher, + final TargetManagement targetManagement, final SoftwareModuleManagement softwareModuleManagement, final DistributionSetManagement distributionSetManagement, final DeploymentManagement deploymentManagement, final TenantConfigurationManagement tenantConfigurationManagement) { @@ -130,7 +133,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { this.systemSecurityContext = systemSecurityContext; this.systemManagement = systemManagement; this.targetManagement = targetManagement; - this.serviceMatcher = serviceMatcher; this.softwareModuleManagement = softwareModuleManagement; this.distributionSetManagement = distributionSetManagement; this.deploymentManagement = deploymentManagement; @@ -146,14 +148,11 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { /** * Method to send a message to a RabbitMQ Exchange after the Distribution set has been assign to a Target. * - * @param assignedEvent the object to be sent. + * @param targetAssignDistributionSetServiceEvent event to be processed */ - @EventListener(classes = TargetAssignDistributionSetEvent.class) - protected void targetAssignDistributionSet(final TargetAssignDistributionSetEvent assignedEvent) { - if (shouldSkip(assignedEvent)) { - return; - } - + @EventListener(classes = TargetAssignDistributionSetServiceEvent.class) + protected void targetAssignDistributionSet(final TargetAssignDistributionSetServiceEvent targetAssignDistributionSetServiceEvent) { + final TargetAssignDistributionSetEvent assignedEvent = targetAssignDistributionSetServiceEvent.getRemoteEvent(); final List filteredTargetList = getTargetsWithoutPendingCancellations(assignedEvent.getActions().keySet()); if (!filteredTargetList.isEmpty()) { @@ -165,16 +164,25 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { /** * Listener for Multi-Action events. * - * @param multiActionEvent the Multi-Action event to be processed + * @param multiActionAssignServiceEvent the Multi-Action event to be processed */ - @EventListener(classes = MultiActionEvent.class) - protected void onMultiAction(final MultiActionEvent multiActionEvent) { - if (shouldSkip(multiActionEvent)) { - return; - } + @EventListener(classes = MultiActionAssignServiceEvent.class) + protected void onMultiActionAssign(final MultiActionAssignServiceEvent multiActionAssignServiceEvent) { + final MultiActionAssignEvent multiActionAssignEvent = multiActionAssignServiceEvent.getRemoteEvent(); + log.debug("MultiActionAssignEvent received for {}", multiActionAssignEvent.getControllerIds()); + sendMultiActionRequestMessages(multiActionAssignEvent.getControllerIds()); + } - log.debug("MultiActionEvent received for {}", multiActionEvent.getControllerIds()); - sendMultiActionRequestMessages(multiActionEvent.getControllerIds()); + /** + * Listener for Multi-Action events. + * + * @param multiActionCancelServiceEvent the Multi-Action event to be processed + */ + @EventListener(classes = MultiActionCancelServiceEvent.class) + protected void onMultiActionCancel(final MultiActionCancelServiceEvent multiActionCancelServiceEvent) { + final MultiActionCancelEvent multiActionCancelEvent = multiActionCancelServiceEvent.getRemoteEvent(); + log.debug("MultiActionCancelEvent received for {}", multiActionCancelEvent.getControllerIds()); + sendMultiActionRequestMessages(multiActionCancelEvent.getControllerIds()); } protected void sendUpdateMessageToTarget( @@ -198,14 +206,11 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { * Method to send a message to a RabbitMQ Exchange after the assignment of * the Distribution set to a Target has been canceled. * - * @param cancelEvent that is to be converted to a DMF message + * @param cancelTargetAssignmentServiceEvent that is to be converted to a DMF message */ - @EventListener(classes = CancelTargetAssignmentEvent.class) - protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentEvent cancelEvent) { - if (shouldSkip(cancelEvent)) { - return; - } - + @EventListener(classes = CancelTargetAssignmentServiceEvent.class) + protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentServiceEvent cancelTargetAssignmentServiceEvent) { + final CancelTargetAssignmentEvent cancelEvent = cancelTargetAssignmentServiceEvent.getRemoteEvent(); final List eventTargets = partitionedParallelExecution( cancelEvent.getActions().keySet(), targetManagement::getByControllerID); @@ -221,19 +226,17 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { /** * Method to send a message to a RabbitMQ Exchange after a Target was deleted. * - * @param deleteEvent the TargetDeletedEvent which holds the necessary data for sending a target delete message. + * @param serviceTargetDeleteEvent the TargetDeletedEvent which holds the necessary data for sending a target delete message. */ - @EventListener(classes = TargetDeletedEvent.class) - protected void targetDelete(final TargetDeletedEvent deleteEvent) { - if (shouldSkip(deleteEvent)) { - return; - } - + @EventListener(classes = TargetDeletedServiceEvent.class) + protected void targetDelete(final TargetDeletedServiceEvent serviceTargetDeleteEvent) { + final TargetDeletedEvent deleteEvent = serviceTargetDeleteEvent.getRemoteEvent(); sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress()); } - @EventListener(classes = TargetAttributesRequestedEvent.class) - protected void targetTriggerUpdateAttributes(final TargetAttributesRequestedEvent updateAttributesEvent) { + @EventListener(classes = TargetAttributesRequestedServiceEvent.class) + protected void targetTriggerUpdateAttributes(final TargetAttributesRequestedServiceEvent serviceTargetUpdateAttributesEvent) { + final TargetAttributesRequestedEvent updateAttributesEvent = serviceTargetUpdateAttributesEvent.getRemoteEvent(); sendUpdateAttributesMessageToTarget( updateAttributesEvent.getTenant(), updateAttributesEvent.getControllerId(), updateAttributesEvent.getTargetAddress()); @@ -252,10 +255,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { IpUtil.createAmqpUri(virtualHost, ping.getMessageProperties().getReplyTo())); } - protected boolean shouldSkip(final RemoteApplicationEvent event) { - return !isFromSelf(event); - } - protected void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId, final URI address) { if (!IpUtil.isAmqpUri(address)) { return; @@ -515,10 +514,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService { return targetAddress == null || !IpUtil.isAmqpUri(URI.create(targetAddress)); } - private boolean isFromSelf(final RemoteApplicationEvent event) { - return serviceMatcher == null || serviceMatcher.isFromSelf(event); - } - private boolean hasPendingCancellations(final Long targetId) { return deploymentManagement.hasPendingCancellations(targetId); } diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java index 0ecbc5402..870d196d8 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java +++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java @@ -45,13 +45,11 @@ import org.springframework.amqp.rabbit.listener.FatalExceptionStrategy; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.bus.ServiceMatcher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.PropertySource; @@ -73,8 +71,6 @@ public class DmfApiConfiguration { private final AmqpDeadletterProperties amqpDeadletterProperties; private final ConnectionFactory rabbitConnectionFactory; - private ServiceMatcher serviceMatcher; - public DmfApiConfiguration( final AmqpProperties amqpProperties, final AmqpDeadletterProperties amqpDeadletterProperties, final ConnectionFactory rabbitConnectionFactory) { @@ -83,11 +79,6 @@ public class DmfApiConfiguration { this.rabbitConnectionFactory = rabbitConnectionFactory; } - @Autowired(required = false) // spring setter injection - public void setServiceMatcher(final ServiceMatcher serviceMatcher) { - this.serviceMatcher = serviceMatcher; - } - @Bean public FatalExceptionStrategy sqlFatalSQLExceptionStrategy(final AmqpProperties amqpProperties) { return new SqlFatalExceptionStrategy(amqpProperties.getFatalSqlExceptionPolicy()); @@ -281,7 +272,7 @@ public class DmfApiConfiguration { final SoftwareModuleManagement softwareModuleManagement, final DeploymentManagement deploymentManagement, final TenantConfigurationManagement tenantConfigurationManagement) { return new AmqpMessageDispatcherService(rabbitTemplate, amqpSenderService, artifactUrlHandler, - systemSecurityContext, systemManagement, targetManagement, serviceMatcher, softwareModuleManagement, distributionSetManagement, + systemSecurityContext, systemManagement, targetManagement, softwareModuleManagement, distributionSetManagement, deploymentManagement, tenantConfigurationManagement); } diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java index 5cb1f1329..ec0963b40 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java +++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java @@ -36,6 +36,10 @@ import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest; import org.eclipse.hawkbit.dmf.json.model.DmfSoftwareModule; import org.eclipse.hawkbit.repository.SystemManagement; import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent; import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent; import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent; @@ -69,8 +73,8 @@ import org.springframework.test.context.TestPropertySource; @ActiveProfiles({ "test" }) @SpringBootTest(classes = { JpaRepositoryConfiguration.class }, webEnvironment = SpringBootTest.WebEnvironment.NONE) @TestPropertySource(properties = { - "spring.main.allow-bean-definition-overriding=true", - "spring.cloud.bus.enabled=true" }) + "org.eclipse.hawkbit.events.remote-enabled=false", + "spring.main.allow-bean-definition-overriding=true" }) class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { private static final String TENANT = "DEFAULT"; @@ -108,7 +112,7 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { when(systemManagement.getTenantMetadataWithoutDetails()).thenReturn(tenantMetaData); amqpMessageDispatcherService = new AmqpMessageDispatcherService(rabbitTemplate, senderService, - artifactUrlHandlerMock, systemSecurityContext, systemManagement, targetManagement, serviceMatcher, + artifactUrlHandlerMock, systemSecurityContext, systemManagement, targetManagement, softwareModuleManagement, distributionSetManagement, deploymentManagement, tenantConfigurationManagement); } @@ -131,7 +135,9 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { final Action action = createAction(createDistributionSet); final TargetAssignDistributionSetEvent targetAssignDistributionSetEvent = new TargetAssignDistributionSetEvent(action); - amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetEvent); + final TargetAssignDistributionSetServiceEvent targetAssignDistributionSetServiceEvent = + new TargetAssignDistributionSetServiceEvent(targetAssignDistributionSetEvent); + amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetServiceEvent); final Message sendMessage = getCaptureAddressEvent(targetAssignDistributionSetEvent); final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = assertDownloadAndInstallMessage(sendMessage, action.getId()); @@ -179,7 +185,8 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { Mockito.when(rabbitTemplate.convertSendAndReceive(any())).thenReturn(receivedList); final TargetAssignDistributionSetEvent targetAssignDistributionSetEvent = new TargetAssignDistributionSetEvent(action); - amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetEvent); + final TargetAssignDistributionSetServiceEvent targetAssignDistributionSetServiceEvent = new TargetAssignDistributionSetServiceEvent(targetAssignDistributionSetEvent); + amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetServiceEvent); final Message sendMessage = getCaptureAddressEvent(targetAssignDistributionSetEvent); final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = assertDownloadAndInstallMessage(sendMessage, action.getId()); @@ -219,8 +226,9 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { final String amqpUri = "amqp://anyhost"; final TargetAttributesRequestedEvent targetAttributesRequestedEvent = new TargetAttributesRequestedEvent( TENANT,1L, Target.class, CONTROLLER_ID, amqpUri); - - amqpMessageDispatcherService.targetTriggerUpdateAttributes(targetAttributesRequestedEvent); + final TargetAttributesRequestedServiceEvent targetAttributesRequestedServiceEvent = + new TargetAttributesRequestedServiceEvent(targetAttributesRequestedEvent); + amqpMessageDispatcherService.targetTriggerUpdateAttributes(targetAttributesRequestedServiceEvent); final Message sendMessage = createArgumentCapture(URI.create(amqpUri)); assertUpdateAttributesMessage(sendMessage); @@ -236,7 +244,9 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { when(action.getTenant()).thenReturn(TENANT); when(action.getTarget()).thenReturn(testTarget); final CancelTargetAssignmentEvent cancelTargetAssignmentDistributionSetEvent = new CancelTargetAssignmentEvent(action); - amqpMessageDispatcherService.targetCancelAssignmentToDistributionSet(cancelTargetAssignmentDistributionSetEvent); + final CancelTargetAssignmentServiceEvent serviceCancelTargetAssignmentDistributionSetEvent = + new CancelTargetAssignmentServiceEvent(cancelTargetAssignmentDistributionSetEvent); + amqpMessageDispatcherService.targetCancelAssignmentToDistributionSet(serviceCancelTargetAssignmentDistributionSetEvent); final Message sendMessage = createArgumentCapture(AMQP_URI); assertCancelMessage(sendMessage); @@ -251,9 +261,10 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { // setup final String amqpUri = "amqp://anyhost"; final TargetDeletedEvent targetDeletedEvent = new TargetDeletedEvent(TENANT, 1L, Target.class, CONTROLLER_ID, amqpUri); + final TargetDeletedServiceEvent targetDeletedServiceEvent = new TargetDeletedServiceEvent(targetDeletedEvent); // test - amqpMessageDispatcherService.targetDelete(targetDeletedEvent); + amqpMessageDispatcherService.targetDelete(targetDeletedServiceEvent); // verify final Message sendMessage = createArgumentCapture(URI.create(amqpUri)); @@ -269,9 +280,10 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { // setup final String noAmqpUri = "http://anyhost"; final TargetDeletedEvent targetDeletedEvent = new TargetDeletedEvent(TENANT, 1L, Target.class, CONTROLLER_ID, noAmqpUri); + final TargetDeletedServiceEvent targetDeletedServiceEvent = new TargetDeletedServiceEvent(targetDeletedEvent); // test - amqpMessageDispatcherService.targetDelete(targetDeletedEvent); + amqpMessageDispatcherService.targetDelete(targetDeletedServiceEvent); // verify Mockito.verifyNoInteractions(senderService); @@ -286,9 +298,10 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest { // setup final String noAmqpUri = null; final TargetDeletedEvent targetDeletedEvent = new TargetDeletedEvent(TENANT, 1L, Target.class, CONTROLLER_ID, noAmqpUri); + final TargetDeletedServiceEvent targetDeletedServiceEvent = new TargetDeletedServiceEvent(targetDeletedEvent); // test - amqpMessageDispatcherService.targetDelete(targetDeletedEvent); + amqpMessageDispatcherService.targetDelete(targetDeletedServiceEvent); // verify Mockito.verifyNoInteractions(senderService); diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java index f98f54e89..8213fd365 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java +++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java @@ -128,7 +128,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 1), + @Expect(type = TargetUpdatedEvent.class, count = 1), @Expect(type = TargetPollEvent.class, count = 2) }) void registerTargetWithName() { final String controllerId = TARGET_PREFIX + "registerTargetWithName"; @@ -148,7 +148,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 2), + @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 2) }) void registerTargetWithAttributes() { final String controllerId = TARGET_PREFIX + "registerTargetWithAttributes"; @@ -171,7 +171,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 3), + @Expect(type = TargetUpdatedEvent.class, count = 3), @Expect(type = TargetPollEvent.class, count = 2) }) void registerTargetWithNameAndAttributes() { final String controllerId = TARGET_PREFIX + "registerTargetWithAttributes"; @@ -404,14 +404,14 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 1), + @Expect(type = ActionUpdatedEvent.class, count = 1), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 9), // implicit lock @Expect(type = TargetAttributesRequestedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 2), + @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 1) }) void finishActionStatus() { final String controllerId = TARGET_PREFIX + "finishActionStatus"; @@ -425,7 +425,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class), + @Expect(type = ActionUpdatedEvent.class), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @@ -445,7 +445,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class), + @Expect(type = ActionUpdatedEvent.class), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @@ -484,7 +484,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 1), + @Expect(type = ActionUpdatedEvent.class, count = 1), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @@ -663,7 +663,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 3), // implicit lock @Expect(type = CancelTargetAssignmentEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 1), + @Expect(type = ActionUpdatedEvent.class, count = 1), @Expect(type = TargetUpdatedEvent.class, count = 1), @Expect(type = TargetPollEvent.class, count = 2) }) void receiveCancelUpdateMessageAfterAssignmentWasCanceled() { @@ -691,14 +691,14 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 1), + @Expect(type = ActionUpdatedEvent.class, count = 1), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 9), // implicit lock @Expect(type = CancelTargetAssignmentEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 1), + @Expect(type = TargetUpdatedEvent.class, count = 1), @Expect(type = TargetPollEvent.class, count = 1) }) void actionNotExists() { final String controllerId = TARGET_PREFIX + "actionNotExists"; @@ -738,7 +738,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), @Expect(type = CancelTargetAssignmentEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 2), + @Expect(type = ActionUpdatedEvent.class, count = 2), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @@ -766,7 +766,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 4), + @Expect(type = TargetUpdatedEvent.class, count = 4), @Expect(type = TargetPollEvent.class, count = 1) }) void updateAttributesWithDifferentUpdateModes() { final String controllerId = TARGET_PREFIX + "updateAttributes"; @@ -794,7 +794,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class), + @Expect(type = TargetUpdatedEvent.class), @Expect(type = TargetPollEvent.class, count = 1) }) void updateAttributesWithNoThingId() { final String controllerId = TARGET_PREFIX + "updateAttributesWithNoThingId"; @@ -822,7 +822,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @Test @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class), + @Expect(type = TargetUpdatedEvent.class), @Expect(type = TargetPollEvent.class, count = 1) }) void updateAttributesWithWrongBody() { // setup @@ -857,14 +857,14 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 1), + @Expect(type = ActionUpdatedEvent.class, count = 1), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 9), // implicit lock @Expect(type = TargetAttributesRequestedEvent.class, count = 1), - @Expect(type = TargetUpdatedEvent.class, count = 2), + @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 1) }) void downloadOnlyAssignmentFinishesActionWhenTargetReportsDownloaded() throws IOException { // create target @@ -893,14 +893,14 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class, count = 2), + @Expect(type = ActionUpdatedEvent.class, count = 2), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), @Expect(type = DistributionSetUpdatedEvent.class, count = 1), // implicit lock @Expect(type = SoftwareModuleUpdatedEvent.class, count = 9), // implicit lock @Expect(type = TargetAttributesRequestedEvent.class, count = 2), - @Expect(type = TargetUpdatedEvent.class, count = 3), + @Expect(type = TargetUpdatedEvent.class, count = 3), @Expect(type = TargetPollEvent.class, count = 1) }) void downloadOnlyAssignmentAllowsActionStatusUpdatesWhenTargetReportsFinishedAndUpdatesInstalledDS() throws IOException { @@ -1098,7 +1098,7 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1), @Expect(type = TargetAssignDistributionSetEvent.class, count = 1), - @Expect(type = ActionUpdatedEvent.class), + @Expect(type = ActionUpdatedEvent.class), @Expect(type = ActionCreatedEvent.class, count = 1), @Expect(type = DistributionSetCreatedEvent.class, count = 1), @Expect(type = SoftwareModuleCreatedEvent.class, count = 3), diff --git a/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java b/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java index b0946cd32..af0a29fd9 100644 --- a/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java +++ b/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java @@ -38,8 +38,8 @@ import org.springframework.test.context.TestPropertySource; // Dirty context is necessary to create a new vhost and recreate all necessary beans after every test class. @DirtiesContext(classMode = ClassMode.AFTER_CLASS) @TestPropertySource(properties = { - "spring.main.allow-bean-definition-overriding=true", - "spring.cloud.bus.enabled=true" }) + "org.eclipse.hawkbit.events.remote-enabled=false", + "spring.main.allow-bean-definition-overriding=true" }) @SuppressWarnings("java:S6813") // constructor injects are not possible for test classes public abstract class AbstractAmqpIntegrationTest extends AbstractIntegrationTest { diff --git a/hawkbit-dmf/hawkbit-dmf-server/README.md b/hawkbit-dmf/hawkbit-dmf-server/README.md index 9d721f5ed..770d6d3b9 100644 --- a/hawkbit-dmf/hawkbit-dmf-server/README.md +++ b/hawkbit-dmf/hawkbit-dmf-server/README.md @@ -14,34 +14,6 @@ Or: ```bash run org.eclipse.hawkbit.app.dmf.DMFStart ``` - -### Clustering (Experimental) -The micro-service instances are configured to communicate via Spring Cloud Bus. You could run multiple instances of any -micro-service but hawkbit-mgmt-server. Management server run some schedulers which shall not run simultaneously - e.g. -auto assignment checker and rollouts executor. To run multiple management server instances you shall do some extensions -of hawkbit to ensure that they wont run schedulers simultaneously or you shall configure all instances but one to do not -run schedulers! - -## Optional Protostuff for Spring cloud bus -The micro-service instances are configured to communicate via Spring Cloud Bus. Optionally, you could -use [Protostuff](https://github.com/protostuff/protostuff) based message payload serialization for improved performance. - -**Note**: If Protostuff is enabled it shall be enabled on all microservices! - -Add/Uncomment to/in your `application.properties` : -```properties -spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff -``` - -Add to your `pom.xml` : -```xml - - io.protostuff - protostuff-core - - - io.protostuff - protostuff-runtime - -``` \ No newline at end of file +# Clustering (Experimental!!!) +## Remote Events between micro-services +[See more information](../../site/content/guides/clustering.md) \ No newline at end of file diff --git a/hawkbit-dmf/hawkbit-dmf-server/src/main/resources/application.properties b/hawkbit-dmf/hawkbit-dmf-server/src/main/resources/application.properties index 60e3d08ad..eb65077a7 100644 --- a/hawkbit-dmf/hawkbit-dmf-server/src/main/resources/application.properties +++ b/hawkbit-dmf/hawkbit-dmf-server/src/main/resources/application.properties @@ -13,7 +13,6 @@ spring.application.name=dmf-server spring.main.allow-bean-definition-overriding=true # Logging configuration -logging.level.org.eclipse.hawkbit.eventbus.DeadEventListener=WARN logging.level.org.springframework.boot.actuate.audit.listener.AuditListener=WARN logging.level.org.hibernate.validator.internal.util.Version=WARN # security Log with hints on potential attacks @@ -35,12 +34,8 @@ hawkbit.lock=inMemory # Disable discovery client of spring-cloud-commons spring.cloud.discovery.enabled=false -# Configure communication between services -endpoints.spring.cloud.bus.refresh.enabled=false -endpoints.spring.cloud.bus.env.enabled=false -spring.cloud.stream.bindings.springCloudBusInput.group=dmf-server -# To use protostuff (for instance fot improved performance) you shall uncomment -# the following two lines and add io.protostuff:protostuff-core and io.protostuff:protostuff-runtime to dependencies -#spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -#spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff +# remote events configuration +spring.config.import=classpath:/hawkbit-events-defaults.properties +# Optional: Use protostuff (if enabled) +# spring.cloud.stream.default.content-type=application/binary+protostuff diff --git a/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/java/org/eclipse/hawkbit/mgmt/rest/resource/MgmtTargetTagResourceTest.java b/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/java/org/eclipse/hawkbit/mgmt/rest/resource/MgmtTargetTagResourceTest.java index c52143449..778d9fe07 100644 --- a/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/java/org/eclipse/hawkbit/mgmt/rest/resource/MgmtTargetTagResourceTest.java +++ b/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/java/org/eclipse/hawkbit/mgmt/rest/resource/MgmtTargetTagResourceTest.java @@ -256,7 +256,7 @@ public class MgmtTargetTagResourceTest extends AbstractManagementApiIntegrationT @Test @ExpectEvents({ @Expect(type = TargetTagCreatedEvent.class, count = 1), - @Expect(type = TargetCreatedEvent.class, count = 5), + @Expect(type = TargetCreatedEvent.class, count = 5), @Expect(type = TargetUpdatedEvent.class, count = 5) }) public void getAssignedTargetsWithPagingLimitRequestParameter() throws Exception { final TargetTag tag = testdataFactory.createTargetTags(1, "").get(0); @@ -280,7 +280,7 @@ public class MgmtTargetTagResourceTest extends AbstractManagementApiIntegrationT @Test @ExpectEvents({ @Expect(type = TargetTagCreatedEvent.class, count = 1), - @Expect(type = TargetCreatedEvent.class, count = 5), + @Expect(type = TargetCreatedEvent.class, count = 5), @Expect(type = TargetUpdatedEvent.class, count = 5) }) public void getAssignedTargetsWithPagingLimitAndOffsetRequestParameter() throws Exception { final TargetTag tag = testdataFactory.createTargetTags(1, "").get(0); diff --git a/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/resources/mgmt-test.properties b/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/resources/mgmt-test.properties index 29803bb9a..140242d3b 100644 --- a/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/resources/mgmt-test.properties +++ b/hawkbit-mgmt/hawkbit-mgmt-resource/src/test/resources/mgmt-test.properties @@ -11,6 +11,4 @@ # Logging START - activate to see request/response details #logging.level.org.eclipse.hawkbit.rest.util.MockMvcResultPrinter=DEBUG # Logging END - -# disable spring cloud bus for tests -spring.cloud.bus.enabled=false \ No newline at end of file +org.eclipse.hawkbit.events.remote-enabled=false diff --git a/hawkbit-mgmt/hawkbit-mgmt-server/README.md b/hawkbit-mgmt/hawkbit-mgmt-server/README.md index c3994a996..4e5b63d25 100644 --- a/hawkbit-mgmt/hawkbit-mgmt-server/README.md +++ b/hawkbit-mgmt/hawkbit-mgmt-server/README.md @@ -24,17 +24,46 @@ run org.eclipse.hawkbit.app.mgmt.MgmtServerStart The Management API can be accessed via http://localhost:8080/rest/v1 The root url http://localhost:8080 will redirect directly to the Swagger Management UI -### Clustering (Experimental!!!) +# Clustering (Experimental!!!) +## Events -The micro-service instances are configured to communicate via Spring Cloud Bus. You could run multiple instances of any -micro-service but hawkbit-mgmt-server. Management server run some schedulers which shall not run simultaneously - e.g. -auto assignment checker and rollouts executor. To run multiple management server instances you shall do some extensions -of hawkbit to ensure that they wont run schedulers simultaneously or you shall configure all instances but one to do not -run schedulers! +Event communication between nodes is based on [Spring Cloud Stream](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/). +There are different [binder implementations](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_binders) +available. The _hawkbit Update Server_ uses RabbitMQ binder. -## Optional Protostuff for Spring cloud bus +You can run multiple instances of any micro-service, including those consuming events. +However, the `hawkbit-mgmt-server` should typically be run as a single instance, as it schedules time-sensitive jobs such as auto-assignment checking and rollout execution. +If multiple management server instances are needed, you must extend hawkBit to ensure that scheduled tasks do not run concurrently. +Alternatively, configure all but one instance to disable scheduler execution. -The micro-service instances are configured to communicate via Spring Cloud Bus. Optionally, you could +## Event Channel Types in Spring Cloud Stream + +Remote events in hawkBit are distributed through **two distinct types of channels**: + +### 1. Fanout Event Channel + +- Every service instance listening to `fanoutEventChannel` receives **a copy of every message**, regardless of instance count. +- Common for events that should be processed by each consumer independently + - In-memory cache updates + - Internal state propagation + - Logging or auditing +- Not recommended for scenarios where only one consumer should process an event (see `serviceEventChannel` for that). + +**Note**: Every instance bound to this channel will get its own copy of the message. + +### 2. Service Event Channel + +The `serviceEventChannel` is used to **ensure exclusive consumption of events** across service instances. +Only **one instance per consumer group** receives and processes each message, which is critical for non-idempotent or resource-sensitive operations. + +- Only one instance in a consumer group receives each message. +- Ideal for external integrations, third-party API calls, or any task that must not be duplicated. +- Load-balanced across instances within the same group. + + +### Optional Protostuff for Spring cloud stream + +The micro-service instances are configured to communicate via Spring Cloud Stream. Optionally, you could use [Protostuff](https://github.com/protostuff/protostuff) based message payload serialization for improved performance. **Note**: If Protostuff is enabled it shall be enabled on all microservices! @@ -42,19 +71,18 @@ use [Protostuff](https://github.com/protostuff/protostuff) based message payload Add/Uncomment to/in your `application.properties` : ```properties -spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff +spring.cloud.stream.default.content-type=application/binary+protostuff ``` Add to your `pom.xml` : ```xml - io.protostuff - protostuff-core + io.protostuff + protostuff-core - io.protostuff - protostuff-runtime + io.protostuff + protostuff-runtime ``` diff --git a/hawkbit-mgmt/hawkbit-mgmt-server/src/main/resources/application.properties b/hawkbit-mgmt/hawkbit-mgmt-server/src/main/resources/application.properties index 5d3f86195..8a03ffc05 100644 --- a/hawkbit-mgmt/hawkbit-mgmt-server/src/main/resources/application.properties +++ b/hawkbit-mgmt/hawkbit-mgmt-server/src/main/resources/application.properties @@ -14,7 +14,6 @@ spring.main.allow-bean-definition-overriding=true spring.port=8080 # Logging configuration -logging.level.org.eclipse.hawkbit.eventbus.DeadEventListener=WARN logging.level.org.springframework.boot.actuate.audit.listener.AuditListener=WARN logging.level.org.hibernate.validator.internal.util.Version=WARN # security Log with hints on potential attacks @@ -43,15 +42,6 @@ hawkbit.server.repository.publish-target-poll-event=false # Disable discovery client of spring-cloud-commons spring.cloud.discovery.enabled=false -# Configure communication between services -endpoints.spring.cloud.bus.refresh.enabled=false -endpoints.spring.cloud.bus.env.enabled=false -spring.cloud.stream.bindings.springCloudBusInput.group=mgmt-server - -# To use protostuff (for instance fot improved performance) you shall uncomment -# the following two lines and add io.protostuff:protostuff-core and io.protostuff:protostuff-runtime to dependencies -#spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -#spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff # Swagger Configuration / https://springdoc.org/v2/#properties springdoc.api-docs.version=openapi_3_0 @@ -61,4 +51,9 @@ springdoc.packages-to-scan=org.eclipse.hawkbit.mgmt springdoc.paths-to-exclude=/system/** springdoc.swagger-ui.enabled=true springdoc.swagger-ui.csrf.enabled=true -springdoc.swagger-ui.doc-expansion=none \ No newline at end of file +springdoc.swagger-ui.doc-expansion=none + +# remote events configuration +spring.config.import=classpath:/hawkbit-events-defaults.properties +# Optional: Use protostuff (if enabled) +# spring.cloud.stream.default.content-type=application/binary+protostuff \ No newline at end of file diff --git a/hawkbit-monolith/hawkbit-update-server/README.md b/hawkbit-monolith/hawkbit-update-server/README.md index c825f279b..bbe8fdc3a 100644 --- a/hawkbit-monolith/hawkbit-update-server/README.md +++ b/hawkbit-monolith/hawkbit-update-server/README.md @@ -21,36 +21,3 @@ run org.eclipse.hawkbit.doc.Start ### Usage The Management API can be accessed via http://localhost:8080/rest/v1 - -## Enable Clustering (experimental) - -Clustering in hawkBit is based on _Spring Cloud Bus_. It is enabled by default in microservice apps and disabled (by default) in the -monolith app. To enable it for monolith app you should set (via environment, system properties or properties files) the following: - -Add to your `pom.xml` : - -```properties -spring.autoconfigure.exclude= -spring.cloud.bus.enabled=true -``` - -Optional as well is the addition of [Protostuff](https://github.com/protostuff/protostuff) based message payload -serialization for improved performance. To enable it set (via environment, system properties or properties files): - -```properties -spring.cloud.stream.bindings.springCloudBusInput.content-type=application/binary+protostuff -spring.cloud.stream.bindings.springCloudBusOutput.content-type=application/binary+protostuff -``` - -and add to your `pom.xml` : - -```xml - - io.protostuff - protostuff-core - - - io.protostuff - protostuff-runtime - -``` diff --git a/hawkbit-monolith/hawkbit-update-server/src/main/resources/application.properties b/hawkbit-monolith/hawkbit-update-server/src/main/resources/application.properties index 406594630..231c7d905 100644 --- a/hawkbit-monolith/hawkbit-update-server/src/main/resources/application.properties +++ b/hawkbit-monolith/hawkbit-update-server/src/main/resources/application.properties @@ -13,7 +13,6 @@ spring.application.name=update-server spring.main.allow-bean-definition-overriding=true # Logging configuration -logging.level.org.eclipse.hawkbit.eventbus.DeadEventListener=WARN logging.level.org.springframework.boot.actuate.audit.listener.AuditListener=WARN logging.level.org.hibernate.validator.internal.util.Version=WARN # security Log with hints on potential attacks @@ -47,9 +46,12 @@ hawkbit.server.repository.publish-target-poll-event=false ## Disable RabbitMQ auto configuration. Comment it to enable RabbitMQ support. spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration -## Configuration Spring Bus (disabled by default) - no cluster support. To enable it, enable RabbitMQ (see above) -## and comment the line (spring.cloud.bus.enabled=false) or set spring.cloud.bus.enabled=true -spring.cloud.bus.enabled=false + +## Uncomment bellow to Enable communication between services (disabled by default) - no cluster support. +# To enable it, enable RabbitMQ (see above) +# and set below 'org.eclipse.hawkbit.events.remote-enabled=true' +org.eclipse.hawkbit.events.remote-enabled=false + ## Disable DMF (by default) - no DMF support. To enable it, enable RabbitMQ (see above) and comment the line ## (hawkbit.dmf.rabbitmq.enabled=false) set hawkbit.dmf.rabbitmq.enabled=true hawkbit.dmf.enabled=false diff --git a/hawkbit-repository/hawkbit-repository-api/pom.xml b/hawkbit-repository/hawkbit-repository-api/pom.xml index 5490eab1f..73b3f5985 100644 --- a/hawkbit-repository/hawkbit-repository-api/pom.xml +++ b/hawkbit-repository/hawkbit-repository-api/pom.xml @@ -35,7 +35,7 @@ org.springframework.cloud - spring-cloud-starter-bus-amqp + spring-cloud-starter-stream-rabbit diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/EventPublisherHolder.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/EventPublisherHolder.java index 4ce5ac11f..33d9cf934 100644 --- a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/EventPublisherHolder.java +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/EventPublisherHolder.java @@ -1,5 +1,5 @@ /** - * Copyright (c) 2015 Bosch Software Innovations GmbH and others + * Copyright (c) 2025 Contributors to the Eclipse Foundation * * This program and the accompanying materials are made * available under the terms of the Eclipse Public License 2.0 @@ -9,63 +9,174 @@ */ package org.eclipse.hawkbit.repository.event; -import lombok.AccessLevel; -import lombok.Getter; -import lombok.NoArgsConstructor; +import java.util.Set; + +import jakarta.annotation.PostConstruct; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; +import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent; +import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent; +import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent; +import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionCancelServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.cloud.bus.BusProperties; -import org.springframework.cloud.bus.ServiceMatcher; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; -/** - * A singleton bean which holds the event publisher and service origin id in order to publish remote application events. - * It can be used in beans not instantiated by spring e.g. JPA entities which cannot be auto-wired. - */ -@NoArgsConstructor(access = AccessLevel.PRIVATE) -@SuppressWarnings("java:S6548") // java:S6548 - singleton holder ensures static access to spring resources in some places +@Slf4j public final class EventPublisherHolder { + @Value("${org.eclipse.hawkbit.events.remote-enabled:true}") + private boolean remoteEventsEnabled; + @Value("${org.eclipse.hawkbit.events.remote.destination:fanoutEventChannel}") + private String fanoutEventChannel; + @Value("${org.eclipse.hawkbit.events.remote-service-enabled:true}") + private boolean remoteServiceEventsEnabled; + @Value("${org.eclipse.hawkbit.events.remote.service.destination:serviceEventChannel}") + private String serviceEventChannel; + + private static final EventPublisherHolder SINGLETON = new EventPublisherHolder(); + private ApplicationEventPublisher delegateEventPublisher; + private StreamBridge streamBridge; - @Getter - private ApplicationEventPublisher eventPublisher; - private ServiceMatcher serviceMatcher; - private BusProperties bus; - - /** - * @return the event publisher holder singleton instance - */ public static EventPublisherHolder getInstance() { return SINGLETON; } - @Autowired // spring setter injection - public void setApplicationEventPublisher(final ApplicationEventPublisher eventPublisher) { - this.eventPublisher = eventPublisher; - } - - @Autowired(required = false) // spring setter injection - public void setServiceMatcher(final ServiceMatcher serviceMatcher) { - this.serviceMatcher = serviceMatcher; - } - - @Autowired(required = false) // spring setter injection - public void setBusProperties(final BusProperties bus) { - this.bus = bus; - } - - /** - * @return the service origin Id coming either from {@link ServiceMatcher} when available or {@link BusProperties} otherwise. - */ - public String getApplicationId() { - String id = null; - if (serviceMatcher != null) { - id = serviceMatcher.getBusId(); + @PostConstruct + private void validateRemoteEventConfig() { + if (remoteEventsEnabled && streamBridge == null) { + throw new IllegalStateException("'org.eclipse.hawkbit.events.remote-enabled' is true but streamBridge is not configured. Check if 'spring-cloud-starter-stream-rabbit' dependency is included."); } - if (id == null && bus != null) { - id = bus.getId(); + } + + public static final Set> SERVICE_EVENTS = Set.of( + TargetCreatedEvent.class, + TargetDeletedEvent.class, + MultiActionAssignEvent.class, + MultiActionCancelEvent.class, + TargetAssignDistributionSetEvent.class, + TargetAttributesRequestedEvent.class, + CancelTargetAssignmentEvent.class + ); + + @Autowired + public void setApplicationEventPublisher(final ApplicationEventPublisher delegate) { + this.delegateEventPublisher = delegate; + } + + @Autowired(required = false) + public void setStreamBridge(final StreamBridge streamBridge) { + this.streamBridge = streamBridge; + } + + public ApplicationEventPublisher getEventPublisher() { + return new RoutingEventPublisher(streamBridge, delegateEventPublisher); + } + + class RoutingEventPublisher implements ApplicationEventPublisher { + + private final StreamBridge streamBridge; + private final ApplicationEventPublisher delegate; + + public RoutingEventPublisher(final StreamBridge streamBridge, final ApplicationEventPublisher delegate) { + this.streamBridge = streamBridge; + this.delegate = delegate; + } + + @Override + public void publishEvent(final Object event) { + routeEvent(event); + } + + @Override + public void publishEvent(final ApplicationEvent event) { + routeEvent(event); + } + + private void routeEvent(Object event) { + if (remoteEventsEnabled && event instanceof AbstractRemoteEvent remoteEvent) { + // send events to remote nodes + publishRemotely(remoteEvent); + } else { + // publish locally + publishLocally(event); + } + } + + private void publishRemotely(final AbstractRemoteEvent remoteEvent) { + streamBridge.send(fanoutEventChannel, remoteEvent); + + // some events need to be processed only by single service replica + // wrap the entity event into a service event and send it to the service channel + if (shouldForwardAsServiceEvent(remoteEvent)) { + final AbstractRemoteEvent serviceEvent = toServiceEvent(remoteEvent); + if (serviceEvent != null) { + log.debug("Publishing Service event: {} to remote channel: {}", serviceEvent, serviceEventChannel); + streamBridge.send(serviceEventChannel, serviceEvent); + } else { + log.error("No Service event created for: {}. Skipping send Service event to Service channel. {}", remoteEvent.getClass(), + serviceEventChannel); + } + } + } + + private void publishLocally(final Object event) { + delegate.publishEvent(event); + + // check if the event should be forwarded as a service event even if it is not a remote event + if (shouldForwardAsServiceEvent(event)) { + final AbstractRemoteEvent serviceEvent = toServiceEvent((AbstractRemoteEvent) event); + if (serviceEvent != null) { + log.debug("Publishing Service event: {} to locally.", serviceEvent); + delegate.publishEvent(serviceEvent); + } else { + log.error("No Service event created for: {}. Skipping send Service event locally.", event.getClass()); + } + } + } + + /** + * Checks if the event should be forwarded as a service event. + * If remote service events are enabled and the event is one of the service events, + * + * @param remoteEvent the event to check whether it should be forwarded as a service event + * @return true if the event should be forwarded as a service event, false otherwise + */ + private boolean shouldForwardAsServiceEvent(final Object remoteEvent) { + return remoteServiceEventsEnabled && SERVICE_EVENTS.contains(remoteEvent.getClass()); + } + + private AbstractRemoteEvent toServiceEvent(final AbstractRemoteEvent event) { + if (event instanceof TargetAssignDistributionSetEvent targetAssignDistributionSetEvent) { + return new TargetAssignDistributionSetServiceEvent(targetAssignDistributionSetEvent); + } else if (event instanceof MultiActionAssignEvent multiActionAssignEvent) { + return new MultiActionAssignServiceEvent(multiActionAssignEvent); + } else if (event instanceof MultiActionCancelEvent multiActionCancelEvent) { + return new MultiActionCancelServiceEvent(multiActionCancelEvent); + } else if (event instanceof CancelTargetAssignmentEvent cancelTargetAssignmentEvent) { + return new CancelTargetAssignmentServiceEvent(cancelTargetAssignmentEvent); + } else if (event instanceof TargetDeletedEvent targetDeletedEvent) { + return new TargetDeletedServiceEvent(targetDeletedEvent); + } else if (event instanceof TargetCreatedEvent targetCreatedEvent) { + return new TargetCreatedServiceEvent(targetCreatedEvent); + } else if (event instanceof TargetAttributesRequestedEvent targetAttributesRequestedEvent) { + return new TargetAttributesRequestedServiceEvent(targetAttributesRequestedEvent); + } + return null; } - // due to a bug (?) in Spring Cloud, we cannot pass null for applicationId - return id == null ? "" : id; } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEvent.java new file mode 100644 index 000000000..cd558577e --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEvent.java @@ -0,0 +1,36 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote; + +import java.util.UUID; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import org.springframework.context.ApplicationEvent; + +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") +@Getter +@EqualsAndHashCode(callSuper = false) +@JsonIgnoreProperties(ignoreUnknown = true) +public abstract class AbstractRemoteEvent extends ApplicationEvent { + + private final String id; + + protected AbstractRemoteEvent() { + this("_empty_default_"); + } + + protected AbstractRemoteEvent(Object source) { + super(source); + this.id = UUID.randomUUID().toString(); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/RemoteTenantAwareEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/RemoteTenantAwareEvent.java index 4bb36cf25..8d494547b 100644 --- a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/RemoteTenantAwareEvent.java +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/RemoteTenantAwareEvent.java @@ -10,15 +10,14 @@ package org.eclipse.hawkbit.repository.event.remote; import java.io.Serial; +import java.util.UUID; import lombok.AccessLevel; import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.NoArgsConstructor; import lombok.ToString; -import org.eclipse.hawkbit.repository.event.EventPublisherHolder; import org.eclipse.hawkbit.repository.event.TenantAwareEvent; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; /** * A distributed tenant aware event. It's the base class of the other @@ -29,19 +28,21 @@ import org.springframework.cloud.bus.event.RemoteApplicationEvent; @Getter @EqualsAndHashCode(callSuper = true) @ToString(callSuper = true) -public class RemoteTenantAwareEvent extends RemoteApplicationEvent implements TenantAwareEvent { +public class RemoteTenantAwareEvent extends AbstractRemoteEvent implements TenantAwareEvent { @Serial private static final long serialVersionUID = 1L; private String tenant; + /** + * Constructor. + * + * @param source the for the remote event. + * @param tenant the tenant + */ public RemoteTenantAwareEvent(final String tenant, final Object source) { - super(source == null ? getApplicationId() : source, getApplicationId(), DEFAULT_DESTINATION_FACTORY.getDestination(null)); + super(source == null ? UUID.randomUUID() : source); this.tenant = tenant; } - - private static String getApplicationId() { - return EventPublisherHolder.getInstance().getApplicationId(); - } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/AbstractServiceRemoteEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/AbstractServiceRemoteEvent.java new file mode 100644 index 000000000..28ff47690 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/AbstractServiceRemoteEvent.java @@ -0,0 +1,25 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import lombok.Getter; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; + +@Getter +public abstract class AbstractServiceRemoteEvent extends AbstractRemoteEvent { + + private final T remoteEvent; + + protected AbstractServiceRemoteEvent(T remoteEvent) { + super(remoteEvent == null ? "_empty_source_" : remoteEvent.getSource()); + this.remoteEvent = remoteEvent; + } + +} diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/CancelTargetAssignmentServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/CancelTargetAssignmentServiceEvent.java new file mode 100644 index 000000000..05087c724 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/CancelTargetAssignmentServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent; + +/** + * Service event for {@link CancelTargetAssignmentEvent}. Event that needs single replica processing + */ +public class CancelTargetAssignmentServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public CancelTargetAssignmentServiceEvent(@JsonProperty("payload") final CancelTargetAssignmentEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionAssignServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionAssignServiceEvent.java new file mode 100644 index 000000000..11275a6c8 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionAssignServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent; + +/** + * Service event for {@link MultiActionAssignEvent}. Event that needs single replica processing + */ +public class MultiActionAssignServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public MultiActionAssignServiceEvent(@JsonProperty("payload") final MultiActionAssignEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionCancelServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionCancelServiceEvent.java new file mode 100644 index 000000000..c8c88e913 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/MultiActionCancelServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent; + +/** + * Service event for {@link MultiActionCancelEvent}. Event that needs single replica processing + */ +public class MultiActionCancelServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public MultiActionCancelServiceEvent(@JsonProperty("payload") final MultiActionCancelEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAssignDistributionSetServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAssignDistributionSetServiceEvent.java new file mode 100644 index 000000000..aba2b1a0d --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAssignDistributionSetServiceEvent.java @@ -0,0 +1,35 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent; + +/** + * Service event for {@link TargetAssignDistributionSetEvent}. Event that needs single replica processing + */ +public class TargetAssignDistributionSetServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + /** + * Constructor. + * + * @param remoteEvent the remote event to group + */ + @JsonCreator + public TargetAssignDistributionSetServiceEvent(@JsonProperty("payload") final TargetAssignDistributionSetEvent remoteEvent) { + super(remoteEvent); + } +} diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAttributesRequestedServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAttributesRequestedServiceEvent.java new file mode 100644 index 000000000..8af95e323 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetAttributesRequestedServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent; + +/** + * Service event for {@link TargetAttributesRequestedEvent}. Event that needs single replica processing + */ +public class TargetAttributesRequestedServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public TargetAttributesRequestedServiceEvent(@JsonProperty("payload") final TargetAttributesRequestedEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetCreatedServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetCreatedServiceEvent.java new file mode 100644 index 000000000..7a692b50c --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetCreatedServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; + +/** + * Service event for {@link TargetCreatedEvent}. Event that needs single replica processing + */ +public class TargetCreatedServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public TargetCreatedServiceEvent(@JsonProperty("payload") final TargetCreatedEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetDeletedServiceEvent.java b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetDeletedServiceEvent.java new file mode 100644 index 000000000..38d9d2707 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/event/remote/service/TargetDeletedServiceEvent.java @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote.service; + +import java.io.Serial; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent; + +/** + * Service event for {@link TargetDeletedEvent}. Event that needs single replica processing + */ +public class TargetDeletedServiceEvent extends AbstractServiceRemoteEvent { + + @Serial + private static final long serialVersionUID = 1L; + + + @JsonCreator + public TargetDeletedServiceEvent(@JsonProperty("payload") final TargetDeletedEvent remoteEvent) { + super(remoteEvent); + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java new file mode 100644 index 000000000..039dfb16c --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java @@ -0,0 +1,24 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.event; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.messaging.converter.MappingJackson2MessageConverter; +import org.springframework.util.MimeType; + +public class EventJacksonMessageConverter extends MappingJackson2MessageConverter { + + public EventJacksonMessageConverter() { + super(new MimeType("application", "remote-event-json")); + ObjectMapper objectMapper = new ObjectMapper(); + EventType.getNamedTypes().forEach(objectMapper::registerSubtypes); + setObjectMapper(objectMapper); + } +} diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverter.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java similarity index 95% rename from hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverter.java rename to hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java index 248aa5e72..0e3aa222e 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverter.java +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java @@ -14,7 +14,7 @@ import io.protostuff.ProtobufIOUtil; import io.protostuff.Schema; import io.protostuff.runtime.RuntimeSchema; import lombok.extern.slf4j.Slf4j; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; @@ -30,20 +30,20 @@ import org.springframework.util.MimeType; * values of {@link EventType}. */ @Slf4j -public class BusProtoStuffMessageConverter extends AbstractMessageConverter { +public class EventProtoStuffMessageConverter extends AbstractMessageConverter { public static final MimeType APPLICATION_BINARY_PROTOSTUFF = new MimeType("application", "binary+protostuff"); /** The length of the class type length of the payload. */ private static final byte EVENT_TYPE_LENGTH = 2; - public BusProtoStuffMessageConverter() { + public EventProtoStuffMessageConverter() { super(APPLICATION_BINARY_PROTOSTUFF); } @Override protected boolean supports(final Class aClass) { - return RemoteApplicationEvent.class.isAssignableFrom(aClass); + return AbstractRemoteEvent.class.isAssignableFrom(aClass); } @Override diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventPublisherConfiguration.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventPublisherConfiguration.java index 45c1c73a5..a40de4e57 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventPublisherConfiguration.java +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventPublisherConfiguration.java @@ -10,23 +10,18 @@ package org.eclipse.hawkbit.event; import java.util.concurrent.Executor; +import java.util.function.Consumer; -import io.protostuff.ProtostuffIOUtil; -import io.protostuff.Schema; import org.eclipse.hawkbit.repository.event.ApplicationEventFilter; -import org.eclipse.hawkbit.repository.event.EventPublisherHolder; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; import org.eclipse.hawkbit.repository.event.remote.RemoteTenantAwareEvent; +import org.eclipse.hawkbit.repository.event.EventPublisherHolder; import org.eclipse.hawkbit.security.SystemSecurityContext; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.bus.BusProperties; -import org.springframework.cloud.bus.ConditionalOnBusEnabled; -import org.springframework.cloud.bus.ServiceMatcher; -import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.PropertySource; @@ -37,12 +32,10 @@ import org.springframework.core.ResolvableType; import org.springframework.messaging.converter.MessageConverter; /** - * Autoconfiguration for the event bus. + * Autoconfiguration for the events. */ @Configuration -@RemoteApplicationEventScan(basePackages = "org.eclipse.hawkbit.repository.event.remote") -@PropertySource("classpath:/hawkbit-eventbus-defaults.properties") -@EnableConfigurationProperties(BusProperties.class) +@PropertySource("classpath:/hawkbit-events-defaults.properties") public class EventPublisherConfiguration { /** @@ -66,7 +59,7 @@ public class EventPublisherConfiguration { * @return the singleton instance of the {@link EventPublisherHolder} */ @Bean - EventPublisherHolder eventBusHolder() { + public EventPublisherHolder eventPublisherHolder() { return EventPublisherHolder.getInstance(); } @@ -84,19 +77,12 @@ public class EventPublisherConfiguration { private final SystemSecurityContext systemSecurityContext; private final ApplicationEventFilter applicationEventFilter; - private ServiceMatcher serviceMatcher; - protected TenantAwareApplicationEventPublisher( final SystemSecurityContext systemSecurityContext, final ApplicationEventFilter applicationEventFilter) { this.systemSecurityContext = systemSecurityContext; this.applicationEventFilter = applicationEventFilter; } - @Autowired(required = false) - public void setServiceMatcher(final ServiceMatcher serviceMatcher) { - this.serviceMatcher = serviceMatcher; - } - /** * Was overridden that not every event has to run within an own tenantAware. */ @@ -106,33 +92,53 @@ public class EventPublisherConfiguration { return; } - if (serviceMatcher == null || !(event instanceof final RemoteTenantAwareEvent remoteEvent)) { - super.multicastEvent(event, eventType); + if (event instanceof final RemoteTenantAwareEvent remoteEvent) { + systemSecurityContext.runAsSystemAsTenant(() -> { + super.multicastEvent(event, eventType); + return null; + }, remoteEvent.getTenant()); return; } - if (serviceMatcher.isFromSelf(remoteEvent)) { - super.multicastEvent(event, eventType); - return; - } - - systemSecurityContext.runAsSystemAsTenant(() -> { - super.multicastEvent(event, eventType); - return null; - }, remoteEvent.getTenant()); + super.multicastEvent(event, eventType); } } - @ConditionalOnBusEnabled - @ConditionalOnClass({ Schema.class, ProtostuffIOUtil.class }) - protected static class BusProtoStuffAutoConfiguration { + @Bean + @ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true") + public Consumer serviceEventConsumer(ApplicationEventPublisher publisher) { + return publisher::publishEvent; + } + + @Bean + @ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true") + public Consumer fanoutEventConsumer(ApplicationEventPublisher publisher) { + return publisher::publishEvent; + } + + @ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true") + @ConditionalOnProperty(name = "spring.cloud.stream.default.content-type", havingValue = "application/binary+stuff") + protected static class EventProtoStuffAutoConfiguration { /** - * @return the protostuff io message converter + * @return the protostuff io message converter for events */ @Bean - public MessageConverter busProtoBufConverter() { - return new BusProtoStuffMessageConverter(); + public MessageConverter eventProtoStuffConverter() { + return new EventProtoStuffMessageConverter(); + } + } + + @ConditionalOnProperty(name = "org.eclipse.hawkbit.events.remote-enabled", havingValue = "true") + @ConditionalOnProperty(name = "spring.cloud.stream.default.content-type", havingValue = "application/remote-event-json") + protected static class EventJacksonAutoConfiguration { + + /** + * @return the Jackson message converter for events + */ + @Bean + public MessageConverter eventJacksonMessageConverter() { + return new EventJacksonMessageConverter(); } } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventType.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventType.java index cbd8cdfd9..600ba91db 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventType.java +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventType.java @@ -9,10 +9,12 @@ */ package org.eclipse.hawkbit.event; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; +import com.fasterxml.jackson.databind.jsontype.NamedType; import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.NoArgsConstructor; @@ -24,6 +26,13 @@ import org.eclipse.hawkbit.repository.event.remote.DistributionSetTypeDeletedEve import org.eclipse.hawkbit.repository.event.remote.DownloadProgressEvent; import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent; import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionCancelServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; import org.eclipse.hawkbit.repository.event.remote.RolloutDeletedEvent; import org.eclipse.hawkbit.repository.event.remote.RolloutGroupDeletedEvent; import org.eclipse.hawkbit.repository.event.remote.RolloutStoppedEvent; @@ -166,6 +175,15 @@ public class EventType { TYPES.put(44, TargetTypeCreatedEvent.class); TYPES.put(45, TargetTypeUpdatedEvent.class); TYPES.put(46, TargetTypeDeletedEvent.class); + + // processing events - start from 1000 to leave room for future db events + TYPES.put(1000, TargetCreatedServiceEvent.class); + TYPES.put(1001, TargetDeletedServiceEvent.class); + TYPES.put(1002, TargetAssignDistributionSetServiceEvent.class); + TYPES.put(1003, TargetAttributesRequestedServiceEvent.class); + TYPES.put(1004, CancelTargetAssignmentServiceEvent.class); + TYPES.put(1005, MultiActionAssignServiceEvent.class); + TYPES.put(1006, MultiActionCancelServiceEvent.class); } /** @@ -197,4 +215,10 @@ public class EventType { public Class getTargetClass() { return TYPES.get(value); } + + public static Collection getNamedTypes() { + return TYPES.entrySet().stream() + .map(e -> new NamedType(e.getValue(), String.valueOf(e.getKey()))) + .toList(); + } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-eventbus-defaults.properties b/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-eventbus-defaults.properties deleted file mode 100644 index 4c7bf9e96..000000000 --- a/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-eventbus-defaults.properties +++ /dev/null @@ -1,21 +0,0 @@ -# -# Copyright (c) 2015 Bosch Software Innovations GmbH and others -# -# This program and the accompanying materials are made -# available under the terms of the Eclipse Public License 2.0 -# which is available at https://www.eclipse.org/legal/epl-2.0/ -# -# SPDX-License-Identifier: EPL-2.0 -# - -# Spring cloud bus and stream -spring.cloud.bus.enabled=true -# Disable Cloud Bus default events -spring.cloud.bus.env.enabled=false -spring.cloud.bus.ack.enabled=false -spring.cloud.bus.trace.enabled=false -spring.cloud.bus.refresh.enabled=false -# Disable Cloud Bus endpoints -management.endpoint.busrefresh.access=none -management.endpoint.busenv.access=none -# Spring cloud bus and stream END \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-events-defaults.properties b/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-events-defaults.properties new file mode 100644 index 000000000..481b5cfc2 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-core/src/main/resources/hawkbit-events-defaults.properties @@ -0,0 +1,47 @@ +# +# Copyright (c) 2015 Bosch Software Innovations GmbH and others +# +# This program and the accompanying materials are made +# available under the terms of the Eclipse Public License 2.0 +# which is available at https://www.eclipse.org/legal/epl-2.0/ +# +# SPDX-License-Identifier: EPL-2.0 +# + +org.eclipse.hawkbit.events.remote-enabled=true + +spring.cloud.function.definition=fanoutEventConsumer;serviceEventConsumer + +spring.cloud.stream.default.content-type=application/remote-event-json +# -- Consumer bindings -- +spring.cloud.stream.bindings.fanoutEventConsumer-in-0.destination=fanoutEventChannel +spring.cloud.stream.bindings.serviceEventConsumer-in-0.destination=serviceEventChannel + +# -- Producer bindings (for StreamBridge) -- +spring.cloud.stream.bindings.fanoutEventChannel.destination=fanoutEventChannel +spring.cloud.stream.bindings.serviceEventChannel.destination=serviceEventChannel + +spring.cloud.stream.bindings.serviceEventConsumer-in-0.group=${spring.application.name} + +# Performance +spring.cloud.stream.rabbit.binder.compressionLevel=0 +spring.cloud.stream.rabbit.bindings.fanoutEventConsumer-in-0.consumer.anonymousGroupPrefix=${spring.application.name}- +spring.cloud.stream.rabbit.bindings.fanoutEventConsumer-in-0.consumer.durableSubscription=false +spring.cloud.stream.rabbit.bindings.fanoutEventConsumer-in-0.consumer.maxConcurrency=1 +spring.cloud.stream.rabbit.bindings.fanoutEventConsumer-in-0.consumer.requeueRejected=false +spring.cloud.stream.rabbit.bindings.fanoutEventConsumer-in-0.consumer.prefetch=100 +spring.cloud.stream.rabbit.bindings.serviceEventConsumer-in-0.consumer.maxConcurrency=1 +spring.cloud.stream.rabbit.bindings.serviceEventConsumer-in-0.consumer.requeueRejected=false +spring.cloud.stream.rabbit.bindings.serviceEventConsumer-in-0.consumer.prefetch=100 + +spring.cloud.stream.rabbit.bindings.fanoutEventChannel.producer.declareExchange=false +spring.cloud.stream.rabbit.bindings.fanoutEventChannel.producer.batchingEnabled=true +spring.cloud.stream.rabbit.bindings.fanoutEventChannel.producer.batchSize=1000 +spring.cloud.stream.rabbit.bindings.fanoutEventChannel.producer.batch-buffer-limit=100000 +spring.cloud.stream.rabbit.bindings.fanoutEventChannel.producer.deliveryMode=NON_PERSISTENT + +spring.cloud.stream.rabbit.bindings.serviceEventChannel.producer.declareExchange=false +spring.cloud.stream.rabbit.bindings.serviceEventChannel.producer.batchingEnabled=true +spring.cloud.stream.rabbit.bindings.serviceEventChannel.producer.batchSize=1000 +spring.cloud.stream.rabbit.bindings.serviceEventChannel.producer.batch-buffer-limit=100000 +spring.cloud.stream.rabbit.bindings.serviceEventChannel.producer.deliveryMode=NON_PERSISTENT \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java new file mode 100644 index 000000000..2a721c834 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java @@ -0,0 +1,77 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.event; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +import java.util.HashMap; + +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; +import org.eclipse.hawkbit.repository.model.Target; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +@ExtendWith(MockitoExtension.class) +class EventJacksonMessageConverterTest { + + private final TestEventJacksonMessageConverter underTest = new TestEventJacksonMessageConverter(); + + @Mock + private Target targetMock; + + @Mock + private Message messageMock; + + @BeforeEach + void before() { + when(targetMock.getId()).thenReturn(1L); + } + + /** + * Verifies that the TargetCreatedEvent can be successfully serialized and deserialized + */ + @Test + void successfullySerializeAndDeserializeEvent() { + final TargetCreatedEvent targetCreatedEvent = new TargetCreatedEvent(targetMock); + // serialize + final Object serializedEvent = underTest.convertToInternal(targetCreatedEvent, + new MessageHeaders(new HashMap<>()), null); + assertThat(serializedEvent).isInstanceOf(byte[].class); + + // deserialize + when(messageMock.getPayload()).thenReturn(serializedEvent); + final Object deserializedEvent = underTest.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); + assertThat(deserializedEvent) + .isInstanceOf(TargetCreatedEvent.class) + .isEqualTo(targetCreatedEvent); + } + + /** + * Test subclass to expose protected methods for testing. + */ + private static class TestEventJacksonMessageConverter extends EventJacksonMessageConverter { + @Override + public Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) { + return super.convertToInternal(payload, headers, conversionHint); + } + + @Override + public Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { + return super.convertFromInternal(message, targetClass, conversionHint); + } + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverterTest.java b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java similarity index 92% rename from hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverterTest.java rename to hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java index d9f4c13c8..260472c2b 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/BusProtoStuffMessageConverterTest.java +++ b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java @@ -16,6 +16,7 @@ import static org.mockito.Mockito.when; import java.io.Serial; import java.util.HashMap; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; import org.eclipse.hawkbit.repository.event.remote.entity.RemoteEntityEvent; import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; import org.eclipse.hawkbit.repository.model.Target; @@ -24,15 +25,14 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConversionException; @ExtendWith(MockitoExtension.class) -class BusProtoStuffMessageConverterTest { +class EventProtoStuffMessageConverterTest { - private final BusProtoStuffMessageConverter underTest = new BusProtoStuffMessageConverter(); + private final EventProtoStuffMessageConverter underTest = new EventProtoStuffMessageConverter(); @Mock private Target targetMock; @@ -58,7 +58,7 @@ class BusProtoStuffMessageConverterTest { // deserialize when(messageMock.getPayload()).thenReturn(serializedEvent); - final Object deserializedEvent = underTest.convertFromInternal(messageMock, RemoteApplicationEvent.class, null); + final Object deserializedEvent = underTest.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); assertThat(deserializedEvent) .isInstanceOf(TargetCreatedEvent.class) .isEqualTo(targetCreatedEvent); diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaTenantConfigurationManagement.java b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaTenantConfigurationManagement.java index 7c82bdde2..60f333cc1 100644 --- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaTenantConfigurationManagement.java +++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaTenantConfigurationManagement.java @@ -30,7 +30,6 @@ import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.eclipse.hawkbit.im.authentication.SpPermission; import org.eclipse.hawkbit.repository.TenantConfigurationManagement; -import org.eclipse.hawkbit.repository.event.remote.RemoteTenantAwareEvent; import org.eclipse.hawkbit.repository.event.remote.TenantConfigurationDeletedEvent; import org.eclipse.hawkbit.repository.event.remote.entity.RemoteEntityEvent; import org.eclipse.hawkbit.repository.exception.InsufficientPermissionException; @@ -58,7 +57,6 @@ import org.springframework.cache.Cache; import org.springframework.cache.CacheManager; import org.springframework.cache.annotation.CacheEvict; import org.springframework.cache.annotation.Cacheable; -import org.springframework.cloud.bus.ServiceMatcher; import org.springframework.context.ApplicationContext; import org.springframework.context.event.EventListener; import org.springframework.core.convert.support.ConfigurableConversionService; @@ -89,9 +87,7 @@ public class JpaTenantConfigurationManagement implements TenantConfigurationMana private final CacheManager cacheManager; private final AfterTransactionCommitExecutor afterCommitExecutor; - private ServiceMatcher serviceMatcher; - - protected JpaTenantConfigurationManagement( + public JpaTenantConfigurationManagement( final TenantConfigurationRepository tenantConfigurationRepository, final TenantConfigurationProperties tenantConfigurationProperties, final CacheManager cacheManager, final AfterTransactionCommitExecutor afterCommitExecutor, @@ -103,11 +99,6 @@ public class JpaTenantConfigurationManagement implements TenantConfigurationMana this.applicationContext = applicationContext; } - @Autowired(required = false) - public void setServiceMatcher(final ServiceMatcher serviceMatcher) { - this.serviceMatcher = serviceMatcher; - } - @Override @CacheEvict(value = "tenantConfiguration", key = "#configurationKeyName") @Transactional @@ -186,10 +177,6 @@ public class JpaTenantConfigurationManagement implements TenantConfigurationMana */ @EventListener public void onTenantConfigurationDeletedEvent(final TenantConfigurationDeletedEvent event) { - if (!shouldProcessRemoteTenantAwareEvent(event)) { - return; - } - evictCacheEntryByKeyIfPresent(event.getConfigKey()); } @@ -200,10 +187,6 @@ public class JpaTenantConfigurationManagement implements TenantConfigurationMana */ @EventListener public void onTenantConfigurationRemoteEntityEvent(final RemoteEntityEvent event) { - if (!shouldProcessRemoteTenantAwareEvent(event)) { - return; - } - event.getEntity().ifPresent(tenantConfiguration -> evictCacheEntryByKeyIfPresent(tenantConfiguration.getKey())); } @@ -391,8 +374,4 @@ public class JpaTenantConfigurationManagement implements TenantConfigurationMana cache.evictIfPresent(key); } } - - private boolean shouldProcessRemoteTenantAwareEvent(final RemoteTenantAwareEvent event) { - return serviceMatcher == null || !serviceMatcher.isFromSelf(event) && serviceMatcher.isForSelf(event); - } } diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/model/JpaAction.java b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/model/JpaAction.java index 7a3016f7e..4b8cbfe81 100644 --- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/model/JpaAction.java +++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/model/JpaAction.java @@ -51,7 +51,6 @@ import org.eclipse.hawkbit.repository.event.remote.entity.ActionUpdatedEvent; import org.eclipse.hawkbit.repository.jpa.utils.MapAttributeConverter; import org.eclipse.hawkbit.repository.model.Action; import org.eclipse.hawkbit.repository.model.ActionStatus; -import org.eclipse.hawkbit.repository.model.BaseEntity; import org.eclipse.hawkbit.repository.model.DistributionSet; import org.eclipse.hawkbit.repository.model.Rollout; import org.eclipse.hawkbit.repository.model.RolloutGroup; diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java index e2a50a5f0..1ff670d1a 100644 --- a/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java +++ b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java @@ -11,49 +11,42 @@ package org.eclipse.hawkbit.repository.event.remote; import static org.junit.jupiter.api.Assertions.fail; -import java.util.LinkedHashMap; import java.util.Map; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import org.eclipse.hawkbit.event.BusProtoStuffMessageConverter; +import org.eclipse.hawkbit.event.EventProtoStuffMessageConverter; import org.eclipse.hawkbit.repository.event.TenantAwareEvent; import org.eclipse.hawkbit.repository.jpa.AbstractJpaIntegrationTest; import org.junit.jupiter.api.BeforeEach; -import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; -import org.springframework.cloud.bus.jackson.BusJacksonAutoConfiguration; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.util.ReflectionTestUtils; -import org.springframework.util.ClassUtils; -import org.springframework.util.MimeType; +import org.springframework.messaging.converter.MappingJackson2MessageConverter; +import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.MimeTypeUtils; /** * Test the remote entity events. */ -@TestPropertySource(properties = { "spring.cloud.bus.enabled=true" }) @SuppressWarnings("java:S6813") // constructor injects are not possible for test classes - public abstract class AbstractRemoteEventTest extends AbstractJpaIntegrationTest { +@Import(AbstractRemoteEventTest.EventProtoStuffTestConfig.class) +public abstract class AbstractRemoteEventTest extends AbstractJpaIntegrationTest { @Autowired - private BusProtoStuffMessageConverter busProtoStuffMessageConverter; + private EventProtoStuffMessageConverter eventProtoStuffMessageConverter; + private AbstractMessageConverter jacksonMessageConverter; @BeforeEach - public void setup() throws Exception { - final BusJacksonAutoConfiguration autoConfiguration = new BusJacksonAutoConfiguration(); - this.jacksonMessageConverter = autoConfiguration.busJsonConverter(null); - ReflectionTestUtils.setField( - jacksonMessageConverter, "packagesToScan", - new String[] { "org.eclipse.hawkbit.repository.event.remote", ClassUtils.getPackageName(RemoteApplicationEvent.class) }); - ((InitializingBean) jacksonMessageConverter).afterPropertiesSet(); + public void setup() { + this.jacksonMessageConverter = new MappingJackson2MessageConverter(); } @SuppressWarnings("unchecked") @@ -65,24 +58,33 @@ import org.springframework.util.MimeTypeUtils; @SuppressWarnings("unchecked") protected T createProtoStuffEvent(final T event) { final Message message = createProtoStuffMessage(event); - return (T) busProtoStuffMessageConverter.fromMessage(message, event.getClass()); + return (T) eventProtoStuffMessageConverter.fromMessage(message, event.getClass()); } private Message createProtoStuffMessage(final TenantAwareEvent event) { - final Map headers = new LinkedHashMap<>(); - headers.put(MessageHeaders.CONTENT_TYPE, BusProtoStuffMessageConverter.APPLICATION_BINARY_PROTOSTUFF); - return busProtoStuffMessageConverter.toMessage(event, new MutableMessageHeaders(headers)); + return eventProtoStuffMessageConverter.toMessage( + event, new MutableMessageHeaders(Map.of(MessageHeaders.CONTENT_TYPE, + EventProtoStuffMessageConverter.APPLICATION_BINARY_PROTOSTUFF)) + ); } private Message createJsonMessage(final Object event) { - final Map headers = new LinkedHashMap<>(); - headers.put(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON); try { - final String json = new ObjectMapper().writeValueAsString(event); - return MessageBuilder.withPayload(json).copyHeaders(headers).build(); - } catch (final JsonProcessingException e) { + String json = new ObjectMapper().writeValueAsString(event); + return MessageBuilder.withPayload(json) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + } catch (JsonProcessingException e) { fail(e.getMessage()); } return null; } + + @TestConfiguration + static class EventProtoStuffTestConfig { + @Bean + public MessageConverter eventProtoBufConverter() { + return new EventProtoStuffMessageConverter(); + } + } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/ServiceEventsTest.java b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/ServiceEventsTest.java new file mode 100644 index 000000000..5a63f0980 --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/ServiceEventsTest.java @@ -0,0 +1,150 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.repository.event.remote; + +import org.eclipse.hawkbit.repository.event.EventPublisherHolder; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; +import org.eclipse.hawkbit.repository.model.Action; +import org.eclipse.hawkbit.repository.model.DistributionSet; +import org.eclipse.hawkbit.repository.model.Target; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.ApplicationEventPublisher; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.*; + +import java.util.List; +import java.util.Set; + +class ServiceEventsTest { + + private StreamBridge streamBridge; + private ApplicationEventPublisher delegate; + private ApplicationEventPublisher publisher; + + @BeforeEach + void setUp() throws IllegalAccessException, NoSuchFieldException { + streamBridge = mock(StreamBridge.class); + delegate = mock(ApplicationEventPublisher.class); + EventPublisherHolder.getInstance().setApplicationEventPublisher(delegate); + EventPublisherHolder.getInstance().setStreamBridge(streamBridge); + publisher = EventPublisherHolder.getInstance().getEventPublisher(); + + // Set up fields via reflection + var remoteEventsEnabledField = EventPublisherHolder.class.getDeclaredField("remoteEventsEnabled"); + remoteEventsEnabledField.setAccessible(true); + remoteEventsEnabledField.set(EventPublisherHolder.getInstance(), true); + + var remoteServiceEventsEnabledField = EventPublisherHolder.class.getDeclaredField("remoteServiceEventsEnabled"); + remoteServiceEventsEnabledField.setAccessible(true); + remoteServiceEventsEnabledField.set(EventPublisherHolder.getInstance(), true); + + var fanoutChannelField = EventPublisherHolder.class.getDeclaredField("fanoutEventChannel"); + fanoutChannelField.setAccessible(true); + fanoutChannelField.set(EventPublisherHolder.getInstance(), "fanout"); + + var groupChannelField = EventPublisherHolder.class.getDeclaredField("serviceEventChannel"); + groupChannelField.setAccessible(true); + groupChannelField.set(EventPublisherHolder.getInstance(), "group"); + } + + @Test + void testExpectedServiceEvents(){ + var expected = Set.of( + TargetAssignDistributionSetEvent.class, + MultiActionAssignEvent.class, + MultiActionCancelEvent.class, + CancelTargetAssignmentEvent.class, + TargetDeletedEvent.class, + TargetCreatedEvent.class, + TargetAttributesRequestedEvent.class + ); + assertEquals(EventPublisherHolder.SERVICE_EVENTS, expected); + } + + @Test + void testProcessingTargetAssignDistributionSetEventIsSent() { + TargetAssignDistributionSetEvent event = new TargetAssignDistributionSetEvent(mockAction()); + + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(TargetAssignDistributionSetServiceEvent.class)); + } + + @Test + void testProcessingTargetCreatedEventIsSent() { + TargetCreatedEvent event = new TargetCreatedEvent(mock(Target.class)); + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(TargetCreatedServiceEvent.class)); + } + + @Test + void testProcessingTargetDeletedEventIsSent() { + TargetDeletedEvent event = new TargetDeletedEvent("testtenant", 1l, Target.class, "testControllerId", "address"); + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(TargetDeletedServiceEvent.class)); + } + + @Test + void testProcessingTargetAttributesRequestedEventIsSent() { + TargetAttributesRequestedEvent event = new TargetAttributesRequestedEvent("testtenant", 1l, Target.class, "testControllerId","address"); + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(TargetAttributesRequestedServiceEvent.class)); + } + + @Test + void testProcessingMultiActionAssignmentEventIsSent() { + MultiActionAssignEvent event = new MultiActionAssignEvent("testtenant", List.of(mockAction())); + + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(MultiActionAssignServiceEvent.class)); + } + + @Test + void testCancelTargetAssignmentEventIsSent() { + CancelTargetAssignmentEvent event = new CancelTargetAssignmentEvent(mockAction()); + + publisher.publishEvent(event); + + verify(streamBridge).send("fanout", event); + verify(streamBridge).send(eq("group"), any(CancelTargetAssignmentServiceEvent.class)); + } + + private Action mockAction() { + final Action actionMock = mock(Action.class); + final Target targetMock = mock(Target.class); + final DistributionSet distributionSetMock = mock(DistributionSet.class); + when(distributionSetMock.getId()).thenReturn(1L); + when(actionMock.getDistributionSet()).thenReturn(distributionSetMock); + when(actionMock.getId()).thenReturn(1l); + when(actionMock.getTenant()).thenReturn("DEFAULT"); + when(actionMock.getTarget()).thenReturn(targetMock); + when(actionMock.getActionType()).thenReturn(Action.ActionType.SOFT); + when(targetMock.getControllerId()).thenReturn("target1"); + return actionMock; + } +} \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/test/resources/jpa-test.properties b/hawkbit-repository/hawkbit-repository-jpa/src/test/resources/jpa-test.properties index 26a433cb5..062f18d79 100644 --- a/hawkbit-repository/hawkbit-repository-jpa/src/test/resources/jpa-test.properties +++ b/hawkbit-repository/hawkbit-repository-jpa/src/test/resources/jpa-test.properties @@ -9,7 +9,6 @@ # ### Debug & Monitor Eclipselink - START - logging.level.org.eclipse.persistence=ERROR ## Uncomment to see the debug of persistence, e.g. to see the generated SQLs #logging.level.org.eclipse.persistence=DEBUG @@ -56,5 +55,5 @@ hawkbit.repository.cluster.lock.refreshOnRemainMS=200 hawkbit.repository.cluster.lock.refreshOnRemainPercent=10 # reduce scheduler tic period to speed up tests hawkbit.repository.cluster.lock.ticPeriodMS=10 -# disable spring cloud bus for tests -spring.cloud.bus.enabled=false \ No newline at end of file + +org.eclipse.hawkbit.events.remote-enabled=false \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/TestConfiguration.java b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/TestConfiguration.java index a8adf0a04..e7d90f6ab 100644 --- a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/TestConfiguration.java +++ b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/TestConfiguration.java @@ -24,7 +24,6 @@ import org.eclipse.hawkbit.repository.artifact.ArtifactRepository; import org.eclipse.hawkbit.repository.artifact.urlhandler.ArtifactUrlHandlerProperties; import org.eclipse.hawkbit.repository.artifact.urlhandler.PropertyBasedArtifactUrlHandler; import org.eclipse.hawkbit.cache.TenantAwareCacheManager; -import org.eclipse.hawkbit.event.BusProtoStuffMessageConverter; import org.eclipse.hawkbit.repository.RolloutApprovalStrategy; import org.eclipse.hawkbit.repository.RolloutStatusCache; import org.eclipse.hawkbit.repository.event.ApplicationEventFilter; @@ -51,7 +50,6 @@ import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cache.caffeine.CaffeineCacheManager; -import org.springframework.cloud.bus.ConditionalOnBusEnabled; import org.springframework.context.ApplicationEvent; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -63,7 +61,6 @@ import org.springframework.core.ResolvableType; import org.springframework.data.domain.AuditorAware; import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; -import org.springframework.messaging.converter.MessageConverter; import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.security.access.hierarchicalroles.RoleHierarchyImpl; import org.springframework.security.concurrent.DelegatingSecurityContextExecutorService; @@ -181,7 +178,7 @@ public class TestConfiguration implements AsyncConfigurer { } @Bean - EventPublisherHolder eventBusHolder() { + EventPublisherHolder eventPublisherHolder() { return EventPublisherHolder.getInstance(); } @@ -208,15 +205,6 @@ public class TestConfiguration implements AsyncConfigurer { return new RolloutTestApprovalStrategy(); } - /** - * @return the protostuff io message converter - */ - @Bean - @ConditionalOnBusEnabled - MessageConverter busProtoBufConverter() { - return new BusProtoStuffMessageConverter(); - } - private static class FilterEnabledApplicationEventPublisher extends SimpleApplicationEventMulticaster { private final ApplicationEventFilter applicationEventFilter; diff --git a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/matcher/EventVerifier.java b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/matcher/EventVerifier.java index 239ea3442..d9ce0cff4 100644 --- a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/matcher/EventVerifier.java +++ b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/matcher/EventVerifier.java @@ -15,8 +15,12 @@ import static org.hamcrest.Matchers.equalTo; import static org.junit.jupiter.api.Assertions.fail; import java.io.Serial; +import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; +import java.util.List; import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -27,10 +31,23 @@ import java.util.stream.Stream; import lombok.extern.slf4j.Slf4j; import org.awaitility.Awaitility; import org.awaitility.core.ConditionTimeoutException; +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; +import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionAssignEvent; +import org.eclipse.hawkbit.repository.event.remote.MultiActionCancelEvent; +import org.eclipse.hawkbit.repository.event.remote.service.CancelTargetAssignmentServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionAssignServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.MultiActionCancelServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAssignDistributionSetServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetAttributesRequestedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetDeletedServiceEvent; import org.eclipse.hawkbit.repository.event.remote.RemoteIdEvent; import org.eclipse.hawkbit.repository.event.remote.RemoteTenantAwareEvent; import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent; -import org.springframework.cloud.bus.event.RemoteApplicationEvent; +import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent; +import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; @@ -86,7 +103,50 @@ public class EventVerifier extends AbstractTestExecutionListener { } private Optional getExpectationsFrom(final Method testMethod) { - return Optional.ofNullable(testMethod.getAnnotation(ExpectEvents.class)).map(ExpectEvents::value); + final Optional expectedEvents = Optional.ofNullable(testMethod.getAnnotation(ExpectEvents.class)).map(ExpectEvents::value); + if (expectedEvents.isPresent()) { + List modifiedEvents = new ArrayList<>(Arrays.asList(expectedEvents.get())); + for (Expect event : expectedEvents.get()) { + final Class type = event.type(); + if (type.isAssignableFrom(TargetCreatedEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(TargetCreatedServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(TargetDeletedEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(TargetDeletedServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(TargetAssignDistributionSetEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(TargetAssignDistributionSetServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(MultiActionAssignEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(MultiActionAssignServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(MultiActionCancelEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(MultiActionCancelServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(TargetAttributesRequestedEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(TargetAttributesRequestedServiceEvent.class, event.count())); + } else if (type.isAssignableFrom(CancelTargetAssignmentEvent.class)) { + modifiedEvents.add(toExpectServiceEvent(CancelTargetAssignmentServiceEvent.class, event.count())); + } + } + return Optional.of(modifiedEvents.toArray(new Expect[0])); + } + return Optional.empty(); + } + + private static Expect toExpectServiceEvent(final Class clazz, final int count) { + return new Expect() { + + @Override + public Class annotationType() { + return Expect.class; + } + + @Override + public Class type() { + return clazz; + } + + @Override + public int count() { + return count; + } + }; } private void beforeTest(final TestContext testContext) { @@ -129,12 +189,12 @@ public class EventVerifier extends AbstractTestExecutionListener { testContext.getApplicationContext().getBean(ApplicationEventMulticaster.class).removeApplicationListener(eventCaptor); } - private static class EventCaptor implements ApplicationListener { + private static class EventCaptor implements ApplicationListener { private final ConcurrentHashMap, Integer> capturedEvents = new ConcurrentHashMap<>(); @Override - public void onApplicationEvent(final RemoteApplicationEvent event) { + public void onApplicationEvent(final AbstractRemoteEvent event) { log.debug("Received event {}", event.getClass().getSimpleName()); if (ResetCounterMarkerEvent.class.isAssignableFrom(event.getClass())) { @@ -170,13 +230,13 @@ public class EventVerifier extends AbstractTestExecutionListener { } } - private static final class ResetCounterMarkerEvent extends RemoteApplicationEvent { + private static final class ResetCounterMarkerEvent extends AbstractRemoteEvent { @Serial private static final long serialVersionUID = 1L; private ResetCounterMarkerEvent() { - super(new Object(), "resetcounter", DEFAULT_DESTINATION_FACTORY.getDestination(null)); + super("event-verifier"); } } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/util/AbstractIntegrationTest.java b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/util/AbstractIntegrationTest.java index 442ea2409..8344d0a00 100644 --- a/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/util/AbstractIntegrationTest.java +++ b/hawkbit-repository/hawkbit-repository-test/src/main/java/org/eclipse/hawkbit/repository/test/util/AbstractIntegrationTest.java @@ -30,6 +30,8 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.FileUtils; import org.awaitility.Awaitility; import org.awaitility.core.ConditionFactory; +import org.eclipse.hawkbit.repository.artifact.ArtifactRepository; +import org.eclipse.hawkbit.repository.artifact.exception.ArtifactStoreException; import org.eclipse.hawkbit.repository.ArtifactManagement; import org.eclipse.hawkbit.repository.ConfirmationManagement; import org.eclipse.hawkbit.repository.ControllerManagement; @@ -52,8 +54,6 @@ import org.eclipse.hawkbit.repository.TargetManagement; import org.eclipse.hawkbit.repository.TargetTagManagement; import org.eclipse.hawkbit.repository.TargetTypeManagement; import org.eclipse.hawkbit.repository.TenantConfigurationManagement; -import org.eclipse.hawkbit.repository.artifact.ArtifactRepository; -import org.eclipse.hawkbit.repository.artifact.exception.ArtifactStoreException; import org.eclipse.hawkbit.repository.exception.EntityNotFoundException; import org.eclipse.hawkbit.repository.model.Action; import org.eclipse.hawkbit.repository.model.Action.ActionType; @@ -78,7 +78,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.bus.ServiceMatcher; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Import; import org.springframework.data.domain.PageRequest; @@ -181,8 +180,6 @@ public abstract class AbstractIntegrationTest { @Autowired protected TestdataFactory testdataFactory; - @Autowired(required = false) - protected ServiceMatcher serviceMatcher; @Autowired protected ApplicationEventPublisher eventPublisher; private static final String ARTIFACT_DIRECTORY = createTempDir().getAbsolutePath() + "/" + randomString(20); diff --git a/hawkbit-repository/hawkbit-repository-test/src/main/resources/hawkbit-test-defaults.properties b/hawkbit-repository/hawkbit-repository-test/src/main/resources/hawkbit-test-defaults.properties index de2d280b2..ba30c46b6 100644 --- a/hawkbit-repository/hawkbit-repository-test/src/main/resources/hawkbit-test-defaults.properties +++ b/hawkbit-repository/hawkbit-repository-test/src/main/resources/hawkbit-test-defaults.properties @@ -78,7 +78,4 @@ hawkbit.server.security.dos.maxTargetsPerAutoAssignment=20 hawkbit.server.security.dos.maxActionsPerTarget=20 # Quota - END -# Properties that are managed by autoconfigure module at runtime and not available during test - END - -# disable spring cloud bus for tests -spring.cloud.bus.enabled=false \ No newline at end of file +# Properties that are managed by autoconfigure module at runtime and not available during test - END \ No newline at end of file diff --git a/site/content/guides/clustering.md b/site/content/guides/clustering.md index 111b9f365..3d9c6f062 100644 --- a/site/content/guides/clustering.md +++ b/site/content/guides/clustering.md @@ -15,18 +15,65 @@ the [hawkBit runtimes's README](https://github.com/eclipse-hawkbit/hawkbit/blob/ ## Events -Event communication between nodes is based on [Spring Cloud Bus](https://cloud.spring.io/spring-cloud-bus/) -and [Spring Cloud Stream](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/). There are -different [binder implementations](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_binders) -available. The _hawkbit Update Server_ uses RabbitMQ binder. Every node gets his own queue to receive cluster events, -the default payload is JSON. -If an event is thrown locally at one node, it will be automatically delivered to all other available nodes via the -Spring Cloud Bus's topic exchange: +Event communication between nodes is based on [Spring Cloud Stream](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/). +There are different [binder implementations](http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_binders) +available. The _hawkbit Update Server_ uses RabbitMQ binder. -![](../../images/eventing-within-cluster.png) +You can run multiple instances of any micro-service, including those consuming events. +However, the `hawkbit-mgmt-server` should typically be run as a single instance, as it schedules time-sensitive jobs such as auto-assignment checking and rollout execution. +If multiple management server instances are needed, you must extend hawkBit to ensure that scheduled tasks do not run concurrently. +Alternatively, configure all but one instance to disable scheduler execution. -Via the ServiceMatcher you can check whether an event happened locally at one node or on a different node. -`serviceMatcher.isFromSelf(event)` +## Event Channel Types in Spring Cloud Stream + +Remote events in hawkBit are distributed through **two distinct types of channels**: + +### 1. Fanout Event Channel + +- Every service instance listening to `fanoutEventChannel` receives **a copy of every message**, regardless of instance count. +- Common for events that should be processed by each consumer independently + - In-memory cache updates + - Internal state propagation + - Logging or auditing +- Not recommended for scenarios where only one consumer should process an event (see `serviceEventChannel` for that). + +**Note**: Every instance bound to this channel will get its own copy of the message. + +### 2. Service Event Channel + +The `serviceEventChannel` is used to **ensure exclusive consumption of events** across service instances. +Only **one instance per consumer group** receives and processes each message, which is critical for non-idempotent or resource-sensitive operations. + +- Only one instance in a consumer group receives each message. +- Ideal for external integrations, third-party API calls, or any task that must not be duplicated. +- Load-balanced across instances within the same group. + + +### Optional Protostuff for Spring cloud stream + +The micro-service instances are configured to communicate via Spring Cloud Stream. Optionally, you could +use [Protostuff](https://github.com/protostuff/protostuff) based message payload serialization for improved performance. + +**Note**: If Protostuff is enabled it shall be enabled on all microservices! + +Add/Uncomment to/in your `application.properties` : + +```properties +spring.cloud.stream.default.content-type=application/binary+protostuff +``` + +Add to your `pom.xml` : + +```xml + + io.protostuff + protostuff-core + + + io.protostuff + protostuff-runtime + +``` ## Caching diff --git a/site/static/images/eventing-within-cluster.png b/site/static/images/eventing-within-cluster.png deleted file mode 100644 index 1baa538d1..000000000 Binary files a/site/static/images/eventing-within-cluster.png and /dev/null differ