@Test public void testPeersVersionNegotiationDowngrade() throws Exception { final HttpServletRequest req = mock(HttpServletRequest.class); doReturn("999").when(req).getHeader(eq(HttpHeaders.PROTOCOL_VERSION)); 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()); assertEquals(new Integer(1), response.getMetadata().getFirst(HttpHeaders.PROTOCOL_VERSION)); }
@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", type = "") } ) @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<>(); if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.entrySet().stream().forEach(entry -> { final PeerDTO peer = new PeerDTO(); final NodeIdentifier nodeId = entry.getKey(); final String siteToSiteAddress = nodeId.getSiteToSiteAddress(); peer.setHostname(siteToSiteAddress == null ? nodeId.getApiAddress() : siteToSiteAddress); peer.setPort(nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort()); peer.setSecure(nodeId.isSiteToSiteSecure()); peer.setFlowFileCount(entry.getValue().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 remoteInputHost = properties.getRemoteInputHost(); String localName; try { localName = InetAddress.getLocalHost().getHostName(); } catch (UnknownHostException e) { if (logger.isDebugEnabled()) { logger.debug("Failed to get local host name using InetAddress.", e); } localName = req.getLocalName(); } peer.setHostname(isEmpty(remoteInputHost) ? localName : remoteInputHost); peer.setPort(properties.getRemoteInputHttpPort()); peer.setSecure(properties.isSiteToSiteSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return clusterContext(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", type = "") } ) @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<>(); if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.entrySet().stream().forEach(entry -> { final PeerDTO peer = new PeerDTO(); final NodeIdentifier nodeId = entry.getKey(); final String siteToSiteAddress = nodeId.getSiteToSiteAddress(); peer.setHostname(siteToSiteAddress == null ? nodeId.getApiAddress() : siteToSiteAddress); peer.setPort(nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort()); peer.setSecure(nodeId.isSiteToSiteSecure()); peer.setFlowFileCount(entry.getValue().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 remoteInputHost = properties.getRemoteInputHost(); String localName; try { localName = InetAddress.getLocalHost().getHostName(); } catch (UnknownHostException e) { if (logger.isDebugEnabled()) { logger.debug("Failed to get local host name using InetAddress.", e); } localName = req.getLocalName(); } peer.setHostname(isEmpty(remoteInputHost) ? localName : remoteInputHost); peer.setPort(properties.getRemoteInputHttpPort()); peer.setSecure(properties.isSiteToSiteSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return clusterContext(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", type = "") } ) @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<>(); if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.entrySet().stream().forEach(entry -> { final PeerDTO peer = new PeerDTO(); final NodeIdentifier nodeId = entry.getKey(); final String siteToSiteAddress = nodeId.getSiteToSiteAddress(); peer.setHostname(siteToSiteAddress == null ? nodeId.getApiAddress() : siteToSiteAddress); peer.setPort(nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort()); peer.setSecure(nodeId.isSiteToSiteSecure()); peer.setFlowFileCount(entry.getValue().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 remoteInputHost = properties.getRemoteInputHost(); String localName; try { localName = InetAddress.getLocalHost().getHostName(); } catch (UnknownHostException e) { if (logger.isDebugEnabled()) { logger.debug("Failed to get local host name using InetAddress.", e); } localName = req.getLocalName(); } peer.setHostname(isEmpty(remoteInputHost) ? localName : remoteInputHost); peer.setPort(properties.getRemoteInputHttpPort()); peer.setSecure(properties.isSiteToSiteSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return clusterContext(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", type = "") } ) @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<>(); if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.entrySet().stream().forEach(entry -> { final PeerDTO peer = new PeerDTO(); final NodeIdentifier nodeId = entry.getKey(); final String siteToSiteAddress = nodeId.getSiteToSiteAddress(); peer.setHostname(siteToSiteAddress == null ? nodeId.getApiAddress() : siteToSiteAddress); peer.setPort(nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort()); peer.setSecure(nodeId.isSiteToSiteSecure()); peer.setFlowFileCount(entry.getValue().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 remoteInputHost = properties.getRemoteInputHost(); String localName; try { localName = InetAddress.getLocalHost().getHostName(); } catch (UnknownHostException e) { if (logger.isDebugEnabled()) { logger.debug("Failed to get local host name using InetAddress.", e); } localName = req.getLocalName(); } peer.setHostname(isEmpty(remoteInputHost) ? localName : remoteInputHost); peer.setPort(properties.getRemoteInputHttpPort()); peer.setSecure(properties.isSiteToSiteSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return clusterContext(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", type = "") } ) @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", type = "") } ) @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", type = "") } ) @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<>(); if (properties.isNode()) { try { final Map<NodeIdentifier, NodeWorkload> clusterWorkload = clusterCoordinator.getClusterWorkload(); clusterWorkload.entrySet().stream().forEach(entry -> { final PeerDTO peer = new PeerDTO(); final NodeIdentifier nodeId = entry.getKey(); final String siteToSiteAddress = nodeId.getSiteToSiteAddress(); peer.setHostname(siteToSiteAddress == null ? nodeId.getApiAddress() : siteToSiteAddress); peer.setPort(nodeId.getSiteToSiteHttpApiPort() == null ? nodeId.getApiPort() : nodeId.getSiteToSiteHttpApiPort()); peer.setSecure(nodeId.isSiteToSiteSecure()); peer.setFlowFileCount(entry.getValue().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 remoteInputHost = properties.getRemoteInputHost(); String localName; try { localName = InetAddress.getLocalHost().getHostName(); } catch (UnknownHostException e) { if (logger.isDebugEnabled()) { logger.debug("Failed to get local host name using InetAddress.", e); } localName = req.getLocalName(); } peer.setHostname(isEmpty(remoteInputHost) ? localName : remoteInputHost); peer.setPort(properties.getRemoteInputHttpPort()); peer.setSecure(properties.isSiteToSiteSecure()); peer.setFlowFileCount(0); peers.add(peer); } final PeersEntity entity = new PeersEntity(); entity.setPeers(peers); return clusterContext(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", type = "") } ) @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", type = "") } ) @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);  }