@Test public void testPeers() throws Exception { final HttpServletRequest req = createCommonHttpServletRequest(); final NiFiServiceFacade serviceFacade = mock(NiFiServiceFacade.class); final SiteToSiteResource resource = getSiteToSiteResource(serviceFacade); final Response response = resource.getPeers(req); PeersEntity resultEntity = (PeersEntity) response.getEntity(); assertEquals(200, response.getStatus()); assertEquals(1, resultEntity.getPeers().size()); final PeerDTO peer = resultEntity.getPeers().iterator().next(); assertEquals(8080, peer.getPort()); }
@GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) public Response getPeers(@Context HttpServletRequest req) { authorizeSiteToSite(); if (!properties.isSiteToSiteHttpEnabled()) { return responseCreator.httpSiteToSiteIsNotEnabledResponse(); } final Integer transportProtocolVersion; try { transportProtocolVersion = negotiateTransportProtocolVersion(req, transportProtocolVersionNegotiator); } catch (BadRequestException e) { return responseCreator.badRequestResponse(e); } final List<PeerDTO> peers = new ArrayList<>(); final PeerDescription source = getSourcePeerDescription(req); final boolean modificationNeeded = peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.HTTP); final Map<String, String> headers = modificationNeeded ? getHttpHeaders(req) : null; if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.forEach((nodeId, workload) -> { final String siteToSiteHostname = nodeId.getSiteToSiteAddress() == null ? nodeId.getApiAddress() : nodeId.getSiteToSiteAddress(); final int siteToSitePort = nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort(); PeerDescription target = new PeerDescription(siteToSiteHostname, siteToSitePort, nodeId.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } final PeerDTO peer = new PeerDTO(); peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(workload.getFlowFileCount()); peers.add(peer); }); } catch (IOException e) { throw new RuntimeException("Failed to retrieve cluster workload due to " + e, e); } } else { final PeerDTO peer = new PeerDTO(); final String siteToSiteHostname = getSiteToSiteHostname(req); PeerDescription target = new PeerDescription(siteToSiteHostname, properties.getRemoteInputHttpPort(), properties.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager)).build(); }
SiteToSiteResource extends ApplicationResource { @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) public Response getPeers(@Context HttpServletRequest req) { authorizeSiteToSite(); if (!properties.isSiteToSiteHttpEnabled()) { return responseCreator.httpSiteToSiteIsNotEnabledResponse(); } final Integer transportProtocolVersion; try { transportProtocolVersion = negotiateTransportProtocolVersion(req, transportProtocolVersionNegotiator); } catch (BadRequestException e) { return responseCreator.badRequestResponse(e); } final List<PeerDTO> peers = new ArrayList<>(); final PeerDescription source = getSourcePeerDescription(req); final boolean modificationNeeded = peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.HTTP); final Map<String, String> headers = modificationNeeded ? getHttpHeaders(req) : null; if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.forEach((nodeId, workload) -> { final String siteToSiteHostname = nodeId.getSiteToSiteAddress() == null ? nodeId.getApiAddress() : nodeId.getSiteToSiteAddress(); final int siteToSitePort = nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort(); PeerDescription target = new PeerDescription(siteToSiteHostname, siteToSitePort, nodeId.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } final PeerDTO peer = new PeerDTO(); peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(workload.getFlowFileCount()); peers.add(peer); }); } catch (IOException e) { throw new RuntimeException("Failed to retrieve cluster workload due to " + e, e); } } else { final PeerDTO peer = new PeerDTO(); final String siteToSiteHostname = getSiteToSiteHostname(req); PeerDescription target = new PeerDescription(siteToSiteHostname, properties.getRemoteInputHttpPort(), properties.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager)).build(); } }
SiteToSiteResource extends ApplicationResource { @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) public Response getPeers(@Context HttpServletRequest req) { authorizeSiteToSite(); if (!properties.isSiteToSiteHttpEnabled()) { return responseCreator.httpSiteToSiteIsNotEnabledResponse(); } final Integer transportProtocolVersion; try { transportProtocolVersion = negotiateTransportProtocolVersion(req, transportProtocolVersionNegotiator); } catch (BadRequestException e) { return responseCreator.badRequestResponse(e); } final List<PeerDTO> peers = new ArrayList<>(); final PeerDescription source = getSourcePeerDescription(req); final boolean modificationNeeded = peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.HTTP); final Map<String, String> headers = modificationNeeded ? getHttpHeaders(req) : null; if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.forEach((nodeId, workload) -> { final String siteToSiteHostname = nodeId.getSiteToSiteAddress() == null ? nodeId.getApiAddress() : nodeId.getSiteToSiteAddress(); final int siteToSitePort = nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort(); PeerDescription target = new PeerDescription(siteToSiteHostname, siteToSitePort, nodeId.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } final PeerDTO peer = new PeerDTO(); peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(workload.getFlowFileCount()); peers.add(peer); }); } catch (IOException e) { throw new RuntimeException("Failed to retrieve cluster workload due to " + e, e); } } else { final PeerDTO peer = new PeerDTO(); final String siteToSiteHostname = getSiteToSiteHostname(req); PeerDescription target = new PeerDescription(siteToSiteHostname, properties.getRemoteInputHttpPort(), properties.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager)).build(); } SiteToSiteResource(final NiFiProperties nifiProperties); }
SiteToSiteResource extends ApplicationResource { @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) public Response getPeers(@Context HttpServletRequest req) { authorizeSiteToSite(); if (!properties.isSiteToSiteHttpEnabled()) { return responseCreator.httpSiteToSiteIsNotEnabledResponse(); } final Integer transportProtocolVersion; try { transportProtocolVersion = negotiateTransportProtocolVersion(req, transportProtocolVersionNegotiator); } catch (BadRequestException e) { return responseCreator.badRequestResponse(e); } final List<PeerDTO> peers = new ArrayList<>(); final PeerDescription source = getSourcePeerDescription(req); final boolean modificationNeeded = peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.HTTP); final Map<String, String> headers = modificationNeeded ? getHttpHeaders(req) : null; if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.forEach((nodeId, workload) -> { final String siteToSiteHostname = nodeId.getSiteToSiteAddress() == null ? nodeId.getApiAddress() : nodeId.getSiteToSiteAddress(); final int siteToSitePort = nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort(); PeerDescription target = new PeerDescription(siteToSiteHostname, siteToSitePort, nodeId.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } final PeerDTO peer = new PeerDTO(); peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(workload.getFlowFileCount()); peers.add(peer); }); } catch (IOException e) { throw new RuntimeException("Failed to retrieve cluster workload due to " + e, e); } } else { final PeerDTO peer = new PeerDTO(); final String siteToSiteHostname = getSiteToSiteHostname(req); PeerDescription target = new PeerDescription(siteToSiteHostname, properties.getRemoteInputHttpPort(), properties.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager)).build(); } SiteToSiteResource(final NiFiProperties nifiProperties); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @ApiOperation( value = "Returns the details about this NiFi necessary to communicate via site to site", response = ControllerEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response getSiteToSiteDetails(@Context HttpServletRequest req); @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response getPeers(@Context HttpServletRequest req); void setServiceFacade(final NiFiServiceFacade serviceFacade); void setAuthorizer(Authorizer authorizer); @Override void setClusterCoordinator(final ClusterCoordinator clusterCoordinator); }
SiteToSiteResource extends ApplicationResource { @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) public Response getPeers(@Context HttpServletRequest req) { authorizeSiteToSite(); if (!properties.isSiteToSiteHttpEnabled()) { return responseCreator.httpSiteToSiteIsNotEnabledResponse(); } final Integer transportProtocolVersion; try { transportProtocolVersion = negotiateTransportProtocolVersion(req, transportProtocolVersionNegotiator); } catch (BadRequestException e) { return responseCreator.badRequestResponse(e); } final List<PeerDTO> peers = new ArrayList<>(); final PeerDescription source = getSourcePeerDescription(req); final boolean modificationNeeded = peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.HTTP); final Map<String, String> headers = modificationNeeded ? getHttpHeaders(req) : null; if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.forEach((nodeId, workload) -> { final String siteToSiteHostname = nodeId.getSiteToSiteAddress() == null ? nodeId.getApiAddress() : nodeId.getSiteToSiteAddress(); final int siteToSitePort = nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort(); PeerDescription target = new PeerDescription(siteToSiteHostname, siteToSitePort, nodeId.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } final PeerDTO peer = new PeerDTO(); peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(workload.getFlowFileCount()); peers.add(peer); }); } catch (IOException e) { throw new RuntimeException("Failed to retrieve cluster workload due to " + e, e); } } else { final PeerDTO peer = new PeerDTO(); final String siteToSiteHostname = getSiteToSiteHostname(req); PeerDescription target = new PeerDescription(siteToSiteHostname, properties.getRemoteInputHttpPort(), properties.isSiteToSiteSecure()); if (modificationNeeded) { target = peerDescriptionModifier.modify(source, target, SiteToSiteTransportProtocol.HTTP, PeerDescriptionModifier.RequestType.Peers, new HashMap<>(headers)); } peer.setHostname(target.getHostname()); peer.setPort(target.getPort()); peer.setSecure(target.isSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager)).build(); } SiteToSiteResource(final NiFiProperties nifiProperties); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @ApiOperation( value = "Returns the details about this NiFi necessary to communicate via site to site", response = ControllerEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response getSiteToSiteDetails(@Context HttpServletRequest req); @GET @Path("/peers") @Consumes(MediaType.WILDCARD) @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML}) @ApiOperation( value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { @Authorization(value = "Read - /site-to-site") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response getPeers(@Context HttpServletRequest req); void setServiceFacade(final NiFiServiceFacade serviceFacade); void setAuthorizer(Authorizer authorizer); @Override void setClusterCoordinator(final ClusterCoordinator clusterCoordinator);  }