You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by mp...@apache.org on 2017/06/07 15:51:39 UTC
[1/3] ambari git commit: AMBARI-21193. Add TopologyChange event to
the server (mpapirkovskyy)
Repository: ambari
Updated Branches:
refs/heads/branch-3.0-perf 5d0962421 -> c2ab4a3ce
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/testing/DBInconsistencyTests.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/testing/DBInconsistencyTests.java b/ambari-server/src/test/java/org/apache/ambari/server/testing/DBInconsistencyTests.java
index 84f0277..930d30c 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/testing/DBInconsistencyTests.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/testing/DBInconsistencyTests.java
@@ -23,6 +23,7 @@ import java.util.List;
import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.OrmTestHelper;
@@ -108,7 +109,7 @@ public class DBInconsistencyTests {
Collection<ServiceComponent> scList = cluster.getService("HDFS").getServiceComponents().values();
Assert.assertNotNull(schList);
- cluster.deleteService("HDFS");
+ cluster.deleteService("HDFS", new DeleteHostComponentStatusMetaData());
List<HostComponentDesiredStateEntity> hostComponentDesiredStateEntities =
hostComponentDesiredStateDAO.findAll();
[2/3] ambari git commit: AMBARI-21193. Add TopologyChange event to
the server (mpapirkovskyy)
Posted by mp...@apache.org.
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/AlertUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/AlertUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/AlertUpdateEvent.java
new file mode 100644
index 0000000..df5ab4d
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/AlertUpdateEvent.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.events;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.ambari.server.state.Alert;
+
+public class AlertUpdateEvent implements AmbariUpdateEvent {
+ private List<Alert> alerts = new ArrayList<>();
+
+ public AlertUpdateEvent(List<Alert> alerts) {
+ this.alerts = alerts;
+ }
+
+ public List<Alert> getAlerts() {
+ return alerts;
+ }
+
+ public void setAlerts(List<Alert> alerts) {
+ this.alerts = alerts;
+ }
+
+ @Override
+ public String getDestination() {
+ return "/events/alerts";
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/AmbariUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/AmbariUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/AmbariUpdateEvent.java
new file mode 100644
index 0000000..62f253b
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/AmbariUpdateEvent.java
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.events;
+
+import java.beans.Transient;
+
+public interface AmbariUpdateEvent {
+
+ @Transient
+ String getDestination();
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/CommandEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/CommandEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/CommandEvent.java
new file mode 100644
index 0000000..cee2fa3
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/CommandEvent.java
@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.events;
+
+public class CommandEvent implements AmbariUpdateEvent {
+ @Override
+ public String getDestination() {
+ return "/command";
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/ConfigsUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/ConfigsUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/ConfigsUpdateEvent.java
new file mode 100644
index 0000000..738badb
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/ConfigsUpdateEvent.java
@@ -0,0 +1,200 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.events;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
+import org.apache.ambari.server.orm.entities.ServiceConfigEntity;
+
+public class ConfigsUpdateEvent implements AmbariUpdateEvent {
+
+ private Long serviceConfigId;
+ private Long clusterId;
+ private String serviceName;
+ private Long groupId;
+ private Long version;
+ private String user;
+ private String note;
+ private List<String> hostNames;
+ private Long createtime;
+ private String groupName;
+ //TODO configs
+
+ private List<ClusterConfig> configs = new ArrayList<>();
+
+ public ConfigsUpdateEvent(ServiceConfigEntity configs, String configGroupName, List<String> hostNames) {
+ this.serviceConfigId = configs.getServiceConfigId();
+ this.clusterId = configs.getClusterEntity().getClusterId();
+ this.serviceName = configs.getServiceName();
+ this.groupId = configs.getGroupId();
+ this.version = configs.getVersion();
+ this.user = configs.getUser();
+ this.note = configs.getNote();
+ this.hostNames = hostNames == null ? null : new ArrayList<>(hostNames);
+ for (ClusterConfigEntity clusterConfigEntity : configs.getClusterConfigEntities()) {
+ this.configs.add(new ClusterConfig(clusterConfigEntity.getClusterId(),
+ clusterConfigEntity.getType(),
+ clusterConfigEntity.getTag(),
+ clusterConfigEntity.getVersion()));
+ }
+ this.createtime = configs.getCreateTimestamp();
+ this.groupName = configGroupName;
+ }
+
+ public Long getServiceConfigId() {
+ return serviceConfigId;
+ }
+
+ public void setServiceConfigId(Long serviceConfigId) {
+ this.serviceConfigId = serviceConfigId;
+ }
+
+ public Long getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(Long clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ public Long getGroupId() {
+ return groupId;
+ }
+
+ public void setGroupId(Long groupId) {
+ this.groupId = groupId;
+ }
+
+ public Long getVersion() {
+ return version;
+ }
+
+ public void setVersion(Long version) {
+ this.version = version;
+ }
+
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ public String getNote() {
+ return note;
+ }
+
+ public void setNote(String note) {
+ this.note = note;
+ }
+
+ public List<String> getHostNames() {
+ return hostNames;
+ }
+
+ public void setHostNames(List<String> hostNames) {
+ this.hostNames = hostNames;
+ }
+
+ public List<ClusterConfig> getConfigs() {
+ return configs;
+ }
+
+ public void setConfigs(List<ClusterConfig> configs) {
+ this.configs = configs;
+ }
+
+ public Long getCreatetime() {
+ return createtime;
+ }
+
+ public void setCreatetime(Long createtime) {
+ this.createtime = createtime;
+ }
+
+ public String getGroupName() {
+ return groupName;
+ }
+
+ public void setGroupName(String groupName) {
+ this.groupName = groupName;
+ }
+
+ @Override
+ public String getDestination() {
+ return "/events/configs";
+ }
+
+ public class ClusterConfig {
+ private Long clusterId;
+ private String type;
+ private String tag;
+ private Long version;
+
+ public ClusterConfig(Long clusterId, String type, String tag, Long version) {
+ this.clusterId = clusterId;
+ this.type = type;
+ this.tag = tag;
+ this.version = version;
+ }
+
+ public Long getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(Long clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public String getTag() {
+ return tag;
+ }
+
+ public void setTag(String tag) {
+ this.tag = tag;
+ }
+
+ public Long getVersion() {
+ return version;
+ }
+
+ public void setVersion(Long version) {
+ this.version = version;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/HostComponentUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/HostComponentUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/HostComponentUpdateEvent.java
new file mode 100644
index 0000000..7500fca
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/HostComponentUpdateEvent.java
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.events;
+
+import org.apache.ambari.server.orm.entities.HostComponentStateEntity;
+import org.apache.ambari.server.state.State;
+
+public class HostComponentUpdateEvent implements AmbariUpdateEvent {
+
+ private Long id;
+ private Long clusterId;
+ private String serviceName;
+ private String hostName;
+ private String componentName;
+ private State currentState;
+
+ public HostComponentUpdateEvent(HostComponentStateEntity stateEntity) {
+ this.id = stateEntity.getId();
+ this.clusterId = stateEntity.getClusterId();
+ this.serviceName = stateEntity.getServiceName();
+ this.hostName = stateEntity.getHostEntity().getHostName();
+ this.currentState = stateEntity.getCurrentState();
+ this.componentName = stateEntity.getComponentName();
+ }
+
+ public Long getId() {
+ return id;
+ }
+
+ public void setId(Long id) {
+ this.id = id;
+ }
+
+ public Long getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(Long clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ public void setHostName(String hostName) {
+ this.hostName = hostName;
+ }
+
+ public String getComponentName() {
+ return componentName;
+ }
+
+ public void setComponentName(String componentName) {
+ this.componentName = componentName;
+ }
+
+ public State getCurrentState() {
+ return currentState;
+ }
+
+ public void setCurrentState(State currentState) {
+ this.currentState = currentState;
+ }
+
+ @Override
+ public String getDestination() {
+ return "/events/hostcomponents";
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/MetadataUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/MetadataUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/MetadataUpdateEvent.java
new file mode 100644
index 0000000..956b73c
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/MetadataUpdateEvent.java
@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.events;
+
+public class MetadataUpdateEvent implements AmbariUpdateEvent {
+ @Override
+ public String getDestination() {
+ return "/events/metadata";
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/RequestUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/RequestUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/RequestUpdateEvent.java
new file mode 100644
index 0000000..682ce50
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/RequestUpdateEvent.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.events;
+
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.controller.internal.CalculatedStatus;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
+import org.apache.ambari.server.orm.entities.RequestEntity;
+import org.apache.ambari.server.topology.TopologyManager;
+
+public class RequestUpdateEvent implements AmbariUpdateEvent {
+
+ private Long clusterId;
+ private Long endTime;
+ private Long requestId;
+ private Double progressPercent;
+ private String requestContext;
+ private HostRoleStatus requestStatus;
+ private Long startTime;
+
+
+ public RequestUpdateEvent(RequestEntity requestEntity, HostRoleCommandDAO hostRoleCommandDAO, TopologyManager topologyManager) {
+ this.clusterId = requestEntity.getClusterId();
+ this.endTime = requestEntity.getEndTime();
+ this.requestId = requestEntity.getRequestId();
+ this.progressPercent = CalculatedStatus.statusFromRequest(hostRoleCommandDAO, topologyManager, requestEntity.getRequestId()).getPercent();
+ this.requestContext = requestEntity.getRequestContext();
+ this.requestStatus = requestEntity.getStatus();
+ this.startTime = requestEntity.getStartTime();
+ }
+
+ public Long getRequestId() {
+ return requestId;
+ }
+
+ public void setRequestId(Long requestId) {
+ this.requestId = requestId;
+ }
+
+ public Long getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(Long clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public String getRequestContext() {
+ return requestContext;
+ }
+
+ public void setRequestContext(String requestContext) {
+ this.requestContext = requestContext;
+ }
+
+ public Long getEndTime() {
+ return endTime;
+ }
+
+ public void setEndTime(Long endTime) {
+ this.endTime = endTime;
+ }
+
+ public Double getProgressPercent() {
+ return progressPercent;
+ }
+
+ public void setProgressPercent(Double progressPercent) {
+ this.progressPercent = progressPercent;
+ }
+
+ public HostRoleStatus getRequestStatus() {
+ return requestStatus;
+ }
+
+ public void setRequestStatus(HostRoleStatus requestStatus) {
+ this.requestStatus = requestStatus;
+ }
+
+ public Long getStartTime() {
+ return startTime;
+ }
+
+ public void setStartTime(Long startTime) {
+ this.startTime = startTime;
+ }
+
+ @Override
+ public String getDestination() {
+ return "/events/requests";
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/TopologyUpdateEvent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/TopologyUpdateEvent.java b/ambari-server/src/main/java/org/apache/ambari/server/events/TopologyUpdateEvent.java
new file mode 100644
index 0000000..9655924
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/TopologyUpdateEvent.java
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.events;
+
+import java.util.Map;
+
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class TopologyUpdateEvent implements AmbariUpdateEvent {
+ @JsonProperty("clustersTopologies")
+ private Map<String, TopologyCluster> clusters;
+
+ private EventType eventType;
+
+ public TopologyUpdateEvent(Map<String, TopologyCluster> clusters, EventType eventType) {
+ this.clusters = clusters;
+ this.eventType = eventType;
+ }
+
+ public Map<String, TopologyCluster> getClusters() {
+ return clusters;
+ }
+
+ public void setClusters(Map<String, TopologyCluster> clusters) {
+ this.clusters = clusters;
+ }
+
+ public EventType getEventType() {
+ return eventType;
+ }
+
+ public void setEventType(EventType eventType) {
+ this.eventType = eventType;
+ }
+
+ @Override
+ public String getDestination() {
+ return "/events/topology";
+ }
+
+ public enum EventType {
+ ADD,
+ DELETE,
+ UPDATE
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertReceivedListener.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertReceivedListener.java b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertReceivedListener.java
index 2faadba..0976fd5 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertReceivedListener.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertReceivedListener.java
@@ -31,8 +31,10 @@ import org.apache.ambari.server.controller.RootServiceResponseFactory.Services;
import org.apache.ambari.server.events.AlertEvent;
import org.apache.ambari.server.events.AlertReceivedEvent;
import org.apache.ambari.server.events.AlertStateChangeEvent;
+import org.apache.ambari.server.events.AlertUpdateEvent;
import org.apache.ambari.server.events.InitialAlertEvent;
import org.apache.ambari.server.events.publishers.AlertEventPublisher;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.orm.RequiresSession;
import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
import org.apache.ambari.server.orm.dao.AlertsDAO;
@@ -88,6 +90,9 @@ public class AlertReceivedListener {
@Inject
Provider<Clusters> m_clusters;
+ @Inject
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
/**
* Used to calculate the maintenance state of new alerts being created.
* Consider the case where you have disabled alerts for a component in MM.
@@ -143,6 +148,7 @@ public class AlertReceivedListener {
List<AlertCurrentEntity> toCreateHistoryAndMerge = new ArrayList<>();
List<AlertEvent> alertEvents = new ArrayList<>(20);
+ List<Alert> updatedAlerts = new ArrayList<>();
for (Alert alert : alerts) {
// jobs that were running when a service/component/host was changed
@@ -339,6 +345,7 @@ public class AlertReceivedListener {
// create the event to fire later
alertEvents.add(new AlertStateChangeEvent(clusterId, alert, current, oldState, oldFirmness));
+ updatedAlerts.add(alert);
}
}
@@ -350,6 +357,9 @@ public class AlertReceivedListener {
for (AlertEvent eventToFire : alertEvents) {
m_alertEventPublisher.publish(eventToFire);
}
+ if (!updatedAlerts.isEmpty()) {
+ stateUpdateEventPublisher.publish(new AlertUpdateEvent(updatedAlerts));
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/requests/StateUpdateListener.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/requests/StateUpdateListener.java b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/requests/StateUpdateListener.java
new file mode 100644
index 0000000..940dc76
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/requests/StateUpdateListener.java
@@ -0,0 +1,52 @@
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.events.listeners.requests;
+
+import org.apache.ambari.server.events.AmbariUpdateEvent;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+
+import com.google.common.eventbus.AllowConcurrentEvents;
+import com.google.common.eventbus.Subscribe;
+import com.google.inject.Injector;
+
+public class StateUpdateListener {
+
+ private final static Logger LOG = LoggerFactory.getLogger(StateUpdateListener.class);
+
+ @Autowired
+ SimpMessagingTemplate simpMessagingTemplate;
+
+ public StateUpdateListener(Injector injector) {
+ StateUpdateEventPublisher stateUpdateEventPublisher =
+ injector.getInstance(StateUpdateEventPublisher.class);
+ stateUpdateEventPublisher.register(this);
+ }
+
+ @Subscribe
+ @AllowConcurrentEvents
+ public void onUpdateEvent(AmbariUpdateEvent event) {
+ LOG.debug("Received status update event {}", event.toString());
+ simpMessagingTemplate.convertAndSend(event.getDestination(), event);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListener.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListener.java b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListener.java
index 93eef36..9eb9f77 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListener.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListener.java
@@ -35,15 +35,19 @@ import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Request;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.controller.internal.CalculatedStatus;
+import org.apache.ambari.server.events.RequestUpdateEvent;
import org.apache.ambari.server.events.TaskCreateEvent;
import org.apache.ambari.server.events.TaskUpdateEvent;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.events.publishers.TaskEventPublisher;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.RequestDAO;
import org.apache.ambari.server.orm.dao.StageDAO;
import org.apache.ambari.server.orm.entities.RequestEntity;
import org.apache.ambari.server.orm.entities.RoleSuccessCriteriaEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.orm.entities.StageEntityPK;
+import org.apache.ambari.server.topology.TopologyManager;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -92,11 +96,23 @@ public class TaskStatusListener {
private RequestDAO requestDAO;
+ private HostRoleCommandDAO hostRoleCommandDAO;
+
+ private TopologyManager topologyManager;
+
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
@Inject
- public TaskStatusListener(TaskEventPublisher taskEventPublisher, StageDAO stageDAO, RequestDAO requestDAO) {
+ public TaskStatusListener(TaskEventPublisher taskEventPublisher, StageDAO stageDAO, RequestDAO requestDAO,
+ StateUpdateEventPublisher stateUpdateEventPublisher,
+ HostRoleCommandDAO hostRoleCommandDAO,
+ TopologyManager topologyManager) {
this.stageDAO = stageDAO;
this.requestDAO = requestDAO;
+ this.stateUpdateEventPublisher = stateUpdateEventPublisher;
+ this.hostRoleCommandDAO = hostRoleCommandDAO;
+ this.topologyManager = topologyManager;
taskEventPublisher.register(this);
}
@@ -263,7 +279,8 @@ public class TaskStatusListener {
ActiveRequest request = activeRequestMap.get(reportedRequestId);
Boolean didStatusChange = updateRequestStatus(reportedRequestId, stagesWithChangedTaskStatus);
if (didStatusChange) {
- requestDAO.updateStatus(reportedRequestId, request.getStatus(), request.getDisplayStatus());
+ RequestEntity updated = requestDAO.updateStatus(reportedRequestId, request.getStatus(), request.getDisplayStatus());
+ stateUpdateEventPublisher.publish(new RequestUpdateEvent(updated, hostRoleCommandDAO, topologyManager));
}
if (request.isCompleted() && isAllTasksCompleted(reportedRequestId)) {
// Request is considered ton have been finished if request status and all of it's tasks status are completed
@@ -606,4 +623,4 @@ public class TaskStatusListener {
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/events/publishers/StateUpdateEventPublisher.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/events/publishers/StateUpdateEventPublisher.java b/ambari-server/src/main/java/org/apache/ambari/server/events/publishers/StateUpdateEventPublisher.java
new file mode 100644
index 0000000..62fe44c
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/events/publishers/StateUpdateEventPublisher.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.events.publishers;
+
+import java.util.concurrent.Executors;
+
+import org.apache.ambari.server.events.AmbariUpdateEvent;
+
+import com.google.common.eventbus.AsyncEventBus;
+import com.google.common.eventbus.EventBus;
+import com.google.inject.Singleton;
+
+@Singleton
+public final class StateUpdateEventPublisher {
+
+ private final EventBus m_eventBus;
+
+ public StateUpdateEventPublisher() {
+ m_eventBus = new AsyncEventBus("ambari-update-bus",
+ Executors.newSingleThreadExecutor());
+ }
+
+ public void publish(AmbariUpdateEvent event) {
+ m_eventBus.post(event);
+ }
+
+ public void register(Object object) {
+ m_eventBus.register(object);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
index 8f16cb2..91d9fd1 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
@@ -192,11 +192,11 @@ public class RequestDAO implements Cleanable {
}
@Transactional
- public void updateStatus(long requestId, HostRoleStatus status, HostRoleStatus displayStatus) {
+ public RequestEntity updateStatus(long requestId, HostRoleStatus status, HostRoleStatus displayStatus) {
RequestEntity requestEntity = findByPK(requestId);
requestEntity.setStatus(status);
requestEntity.setDisplayStatus(displayStatus);
- merge(requestEntity);
+ return merge(requestEntity);
}
@Transactional
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ResourceEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ResourceEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ResourceEntity.java
index 16c1660..53146b1 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ResourceEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ResourceEntity.java
@@ -34,8 +34,6 @@ import javax.persistence.OneToMany;
import javax.persistence.Table;
import javax.persistence.TableGenerator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-
/**
* Represents a resource.
*/
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
index 49fc8c0..3e04a87 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
@@ -26,6 +26,7 @@ import java.util.Set;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.controller.ClusterResponse;
import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
@@ -501,7 +502,7 @@ public interface Cluster {
* @param serviceName
* @throws AmbariException
*/
- void deleteService(String serviceName) throws AmbariException;
+ void deleteService(String serviceName, DeleteHostComponentStatusMetaData deleteMetaData) throws AmbariException;
/**
* Gets if the cluster can be deleted
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/Service.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/Service.java b/ambari-server/src/main/java/org/apache/ambari/server/state/Service.java
index 5964e33..1f8c18e 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/Service.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/Service.java
@@ -22,6 +22,7 @@ import java.util.Map;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.controller.ServiceResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
public interface Service {
@@ -82,14 +83,14 @@ public interface Service {
*/
boolean canBeRemoved();
- void deleteAllComponents() throws AmbariException;
+ void deleteAllComponents(DeleteHostComponentStatusMetaData deleteMetaData);
- void deleteServiceComponent(String componentName)
+ void deleteServiceComponent(String componentName, DeleteHostComponentStatusMetaData deleteMetaData)
throws AmbariException;
boolean isClientOnlyService();
- void delete() throws AmbariException;
+ void delete(DeleteHostComponentStatusMetaData deleteMetaData);
/**
* Sets the maintenance state for the service
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponent.java b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponent.java
index e93ab9a..3995c23 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponent.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponent.java
@@ -22,6 +22,7 @@ import java.util.Map;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.controller.ServiceComponentResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
public interface ServiceComponent {
@@ -88,13 +89,13 @@ public interface ServiceComponent {
boolean canBeRemoved();
- void deleteAllServiceComponentHosts() throws AmbariException;
+ void deleteAllServiceComponentHosts(DeleteHostComponentStatusMetaData deleteMetaData) throws AmbariException;
- void deleteServiceComponentHosts(String hostname)
+ void deleteServiceComponentHosts(String hostname, DeleteHostComponentStatusMetaData deleteMetaData)
throws AmbariException;
ServiceComponentHost addServiceComponentHost(
String hostName) throws AmbariException;
- void delete() throws AmbariException;
+ void delete(DeleteHostComponentStatusMetaData deleteMetaData);
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java
index 104e456..8e6d33b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.controller.ServiceComponentHostResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
@@ -193,7 +194,7 @@ public interface ServiceComponentHost {
boolean canBeRemoved();
- void delete() throws AmbariException;
+ void delete(DeleteHostComponentStatusMetaData deleteMetaData);
/**
* Updates the tags that have been recognized by a START action.
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentImpl.java
index f6ddc6d..dbc7464 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentImpl.java
@@ -31,6 +31,7 @@ import org.apache.ambari.server.ObjectNotFoundException;
import org.apache.ambari.server.ServiceComponentHostNotFoundException;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ServiceComponentResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.ServiceComponentRecoveryChangedEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.dao.ClusterServiceDAO;
@@ -508,7 +509,7 @@ public class ServiceComponentImpl implements ServiceComponent {
@Override
@Transactional
- public void deleteAllServiceComponentHosts() throws AmbariException {
+ public void deleteAllServiceComponentHosts(DeleteHostComponentStatusMetaData deleteMetaData) {
readWriteLock.writeLock().lock();
try {
LOG.info("Deleting all servicecomponenthosts for component" + ", clusterName="
@@ -516,15 +517,16 @@ public class ServiceComponentImpl implements ServiceComponent {
+ ", recoveryEnabled=" + isRecoveryEnabled());
for (ServiceComponentHost sch : hostComponents.values()) {
if (!sch.canBeRemoved()) {
- throw new AmbariException("Found non removable hostcomponent " + " when trying to delete"
+ deleteMetaData.setAmbariException(new AmbariException("Found non removable hostcomponent " + " when trying to delete"
+ " all hostcomponents from servicecomponent" + ", clusterName=" + getClusterName()
+ ", serviceName=" + getServiceName() + ", componentName=" + getName()
- + ", recoveryEnabled=" + isRecoveryEnabled() + ", hostname=" + sch.getHostName());
+ + ", recoveryEnabled=" + isRecoveryEnabled() + ", hostname=" + sch.getHostName()));
+ return;
}
}
for (ServiceComponentHost serviceComponentHost : hostComponents.values()) {
- serviceComponentHost.delete();
+ serviceComponentHost.delete(deleteMetaData);
}
hostComponents.clear();
@@ -534,7 +536,7 @@ public class ServiceComponentImpl implements ServiceComponent {
}
@Override
- public void deleteServiceComponentHosts(String hostname) throws AmbariException {
+ public void deleteServiceComponentHosts(String hostname, DeleteHostComponentStatusMetaData deleteMetaData) throws AmbariException {
readWriteLock.writeLock().lock();
try {
ServiceComponentHost sch = getServiceComponentHost(hostname);
@@ -549,7 +551,7 @@ public class ServiceComponentImpl implements ServiceComponent {
+ ", recoveryEnabled=" + isRecoveryEnabled()
+ ", hostname=" + sch.getHostName());
}
- sch.delete();
+ sch.delete(deleteMetaData);
hostComponents.remove(hostname);
} finally {
@@ -559,10 +561,13 @@ public class ServiceComponentImpl implements ServiceComponent {
@Override
@Transactional
- public void delete() throws AmbariException {
+ public void delete(DeleteHostComponentStatusMetaData deleteMetaData) {
readWriteLock.writeLock().lock();
try {
- deleteAllServiceComponentHosts();
+ deleteAllServiceComponentHosts(deleteMetaData);
+ if (deleteMetaData.getAmbariException() != null) {
+ return;
+ }
ServiceComponentDesiredStateEntity desiredStateEntity = serviceComponentDesiredStateDAO.findById(
desiredStateEntityId);
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceImpl.java
index a0c0db1..f9202e0 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/ServiceImpl.java
@@ -31,6 +31,7 @@ import org.apache.ambari.server.ObjectNotFoundException;
import org.apache.ambari.server.ServiceComponentNotFoundException;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ServiceResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.MaintenanceModeEvent;
import org.apache.ambari.server.events.ServiceInstalledEvent;
import org.apache.ambari.server.events.ServiceRemovedEvent;
@@ -516,7 +517,7 @@ public class ServiceImpl implements Service {
@Override
@Transactional
- public void deleteAllComponents() throws AmbariException {
+ public void deleteAllComponents(DeleteHostComponentStatusMetaData deleteMetaData) {
lock.lock();
try {
LOG.info("Deleting all components for service" + ", clusterName=" + cluster.getClusterName()
@@ -524,14 +525,18 @@ public class ServiceImpl implements Service {
// FIXME check dependencies from meta layer
for (ServiceComponent component : components.values()) {
if (!component.canBeRemoved()) {
- throw new AmbariException("Found non removable component when trying to"
+ deleteMetaData.setAmbariException(new AmbariException("Found non removable component when trying to"
+ " delete all components from service" + ", clusterName=" + cluster.getClusterName()
- + ", serviceName=" + getName() + ", componentName=" + component.getName());
+ + ", serviceName=" + getName() + ", componentName=" + component.getName()));
+ return;
}
}
for (ServiceComponent serviceComponent : components.values()) {
- serviceComponent.delete();
+ serviceComponent.delete(deleteMetaData);
+ if (deleteMetaData.getAmbariException() != null) {
+ return;
+ }
}
components.clear();
@@ -541,7 +546,7 @@ public class ServiceImpl implements Service {
}
@Override
- public void deleteServiceComponent(String componentName)
+ public void deleteServiceComponent(String componentName, DeleteHostComponentStatusMetaData deleteMetaData)
throws AmbariException {
lock.lock();
try {
@@ -556,7 +561,7 @@ public class ServiceImpl implements Service {
+ ", componentName=" + componentName);
}
- component.delete();
+ component.delete(deleteMetaData);
components.remove(componentName);
} finally {
lock.unlock();
@@ -570,11 +575,19 @@ public class ServiceImpl implements Service {
@Override
@Transactional
- public void delete() throws AmbariException {
- deleteAllComponents();
- deleteAllServiceConfigs();
+ public void delete(DeleteHostComponentStatusMetaData deleteMetaData) {
+ deleteAllComponents(deleteMetaData);
+ if (deleteMetaData.getAmbariException() != null) {
+ return;
+ }
+ try {
+ deleteAllServiceConfigs();
- removeEntities();
+ removeEntities();
+ } catch (AmbariException e) {
+ deleteMetaData.setAmbariException(e);
+ return;
+ }
// publish the service removed event
StackId stackId = cluster.getDesiredStackVersion();
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
index b86c5cd..f41c620 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
@@ -37,6 +37,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.persistence.EntityManager;
@@ -61,14 +62,17 @@ import org.apache.ambari.server.controller.ConfigurationResponse;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.controller.RootServiceResponseFactory.Services;
import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.controller.internal.UpgradeResourceProvider;
import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.events.ClusterEvent;
+import org.apache.ambari.server.events.ConfigsUpdateEvent;
import org.apache.ambari.server.events.jpa.EntityManagerCacheInvalidationEvent;
import org.apache.ambari.server.events.jpa.JPAEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.events.publishers.JPAEventPublisher;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.logging.LockFactory;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.metadata.RoleCommandOrderProvider;
@@ -145,8 +149,10 @@ import org.apache.ambari.server.state.scheduler.RequestExecution;
import org.apache.ambari.server.state.scheduler.RequestExecutionFactory;
import org.apache.ambari.server.state.stack.upgrade.Direction;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostSummary;
+import org.apache.ambari.server.topology.TopologyDeleteFormer;
import org.apache.ambari.server.topology.TopologyRequest;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -292,6 +298,9 @@ public class ClusterImpl implements Cluster {
@Inject
private TopologyRequestDAO topologyRequestDAO;
+ @Inject
+ private TopologyDeleteFormer topologyDeleteFormer;
+
/**
* Data access object used for looking up stacks from the database.
*/
@@ -327,6 +336,9 @@ public class ClusterImpl implements Cluster {
@Inject
private UpgradeContextFactory upgradeContextFactory;
+ @Inject
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
/**
* A simple cache for looking up {@code cluster-env} properties for a cluster.
* This map is changed whenever {{cluster-env}} is changed and we receive a
@@ -2007,9 +2019,12 @@ public class ClusterImpl implements Cluster {
}
}
+ DeleteHostComponentStatusMetaData deleteMetaData = new DeleteHostComponentStatusMetaData();
for (Service service : services.values()) {
- deleteService(service);
+ deleteService(service, deleteMetaData);
+ topologyDeleteFormer.processDeleteMetaDataException(deleteMetaData);
}
+ topologyDeleteFormer.processDeleteCluster(Long.toString(getClusterId()));
services.clear();
} finally {
clusterGlobalLock.writeLock().unlock();
@@ -2017,7 +2032,7 @@ public class ClusterImpl implements Cluster {
}
@Override
- public void deleteService(String serviceName)
+ public void deleteService(String serviceName, DeleteHostComponentStatusMetaData deleteMetaData)
throws AmbariException {
clusterGlobalLock.writeLock().lock();
try {
@@ -2026,11 +2041,12 @@ public class ClusterImpl implements Cluster {
+ getClusterName() + ", serviceName=" + service.getName());
// FIXME check dependencies from meta layer
if (!service.canBeRemoved()) {
- throw new AmbariException("Could not delete service from cluster"
- + ", clusterName=" + getClusterName()
- + ", serviceName=" + service.getName());
+ deleteMetaData.setAmbariException(new AmbariException("Could not delete service from cluster"
+ + ", clusterName=" + getClusterName()
+ + ", serviceName=" + service.getName()));
+ return;
}
- deleteService(service);
+ deleteService(service, deleteMetaData);
services.remove(serviceName);
} finally {
@@ -2048,10 +2064,13 @@ public class ClusterImpl implements Cluster {
* @throws AmbariException
* @see ServiceComponentHost
*/
- private void deleteService(Service service) throws AmbariException {
+ private void deleteService(Service service, DeleteHostComponentStatusMetaData deleteMetaData) {
final String serviceName = service.getName();
- service.delete();
+ service.delete(deleteMetaData);
+ if (deleteMetaData.getAmbariException() != null) {
+ return;
+ }
serviceComponentHosts.remove(serviceName);
@@ -2296,10 +2315,17 @@ public class ClusterImpl implements Cluster {
serviceConfigEntity.setStack(clusterEntity.getDesiredStack());
serviceConfigDAO.create(serviceConfigEntity);
+ List<String> groupHostNames = null;
if (configGroup != null) {
+ if (MapUtils.isNotEmpty(configGroup.getHosts())) {
+ groupHostNames = configGroup.getHosts().entrySet().stream().map(h -> h.getValue().getHostName())
+ .collect(Collectors.toList());
+ }
serviceConfigEntity.setHostIds(new ArrayList<>(configGroup.getHosts().keySet()));
serviceConfigEntity = serviceConfigDAO.merge(serviceConfigEntity);
}
+ stateUpdateEventPublisher.publish(new ConfigsUpdateEvent(serviceConfigEntity,
+ configGroup == null ? null : configGroup.getName(), groupHostNames));
} finally {
clusterGlobalLock.writeLock().unlock();
}
@@ -2556,6 +2582,7 @@ public class ClusterImpl implements Cluster {
throw new ObjectNotFoundException("Service config version with serviceName={} and version={} not found");
}
+ String configGroupName = null;
// disable all configs related to service
if (serviceConfigEntity.getGroupId() == null) {
Collection<String> configTypes = serviceConfigTypes.get(serviceName);
@@ -2573,6 +2600,7 @@ public class ClusterImpl implements Cluster {
Long configGroupId = serviceConfigEntity.getGroupId();
ConfigGroup configGroup = clusterConfigGroups.get(configGroupId);
if (configGroup != null) {
+ configGroupName = configGroup.getName();
Map<String, Config> groupDesiredConfigs = new HashMap<>();
for (ClusterConfigEntity entity : serviceConfigEntity.getClusterConfigEntities()) {
Config config = allConfigs.get(entity.getType()).get(entity.getTag());
@@ -2614,7 +2642,15 @@ public class ClusterImpl implements Cluster {
serviceConfigEntityClone.setNote(serviceConfigVersionNote);
serviceConfigEntityClone.setVersion(nextServiceConfigVersion);
+ List<String> groupHostNames = null;
+ if (CollectionUtils.isNotEmpty(serviceConfigEntity.getHostIds())) {
+ groupHostNames = getHosts().stream()
+ .filter(h -> serviceConfigEntity.getHostIds().contains(h.getHostId()))
+ .map(h -> h.getHostName()).collect(Collectors.toList());
+ }
+
serviceConfigDAO.create(serviceConfigEntityClone);
+ stateUpdateEventPublisher.publish(new ConfigsUpdateEvent(serviceConfigEntityClone, configGroupName, groupHostNames));
return convertToServiceConfigVersionResponse(serviceConfigEntityClone);
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
index f1e93ac..6ba9ff7 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
@@ -21,6 +21,7 @@ package org.apache.ambari.server.state.svccomphost;
import java.text.MessageFormat;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -32,13 +33,19 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.agent.AlertDefinitionCommand;
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+import org.apache.ambari.server.agent.stomp.dto.TopologyComponent;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ServiceComponentHostResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.AlertHashInvalidationEvent;
+import org.apache.ambari.server.events.HostComponentUpdateEvent;
import org.apache.ambari.server.events.MaintenanceModeEvent;
import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
+import org.apache.ambari.server.events.TopologyUpdateEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.orm.dao.HostComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.HostComponentStateDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
@@ -121,6 +128,9 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
@Inject
private RepositoryVersionHelper repositoryVersionHelper;
+ @Inject
+ StateUpdateEventPublisher stateUpdateEventPublisher;
+
/**
* Used for creating commands to send to the agents when alert definitions are
* added as the result of a service install.
@@ -895,6 +905,17 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
if (stateEntity != null) {
stateEntity.setVersion(version);
stateEntity = hostComponentStateDAO.merge(stateEntity);
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
+ topologyUpdates.put(Long.toString(getClusterId()), new TopologyCluster());
+ Long hostId = getHost().getHostId();
+ topologyUpdates.get(Long.toString(getClusterId())).addTopologyComponent(TopologyComponent.newBuilder()
+ .setComponentName(getServiceComponentName())
+ .setVersion(stateEntity.getVersion())
+ .setHostIds(new HashSet<>(Collections.singletonList(hostId)))
+ .build());
+ TopologyUpdateEvent hostComponentVersionUpdate = new TopologyUpdateEvent(topologyUpdates,
+ TopologyUpdateEvent.EventType.UPDATE);
+ stateUpdateEventPublisher.publish(hostComponentVersionUpdate);
} else {
LOG.warn("Setting a member on an entity object that may have been "
+ "previously deleted, serviceName = " + getServiceName() + ", " + "componentName = "
@@ -1007,8 +1028,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
try {
stateMachine.doTransition(event.getType(), event);
HostComponentStateEntity stateEntity = getStateEntity();
+ boolean statusUpdated = !stateEntity.getCurrentState().equals(stateMachine.getCurrentState());
stateEntity.setCurrentState(stateMachine.getCurrentState());
stateEntity = hostComponentStateDAO.merge(stateEntity);
+ if (statusUpdated) {
+ stateUpdateEventPublisher.publish(new HostComponentUpdateEvent(stateEntity));
+ }
// TODO Audit logs
} catch (InvalidStateTransitionException e) {
LOG.error("Can't handle ServiceComponentHostEvent event at"
@@ -1359,10 +1384,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
}
@Override
- public void delete() {
+ public void delete(DeleteHostComponentStatusMetaData deleteMetaData) {
boolean fireRemovalEvent = false;
writeLock.lock();
+ String version = getVersion();
try {
removeEntities();
fireRemovalEvent = true;
@@ -1390,6 +1416,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
hostName, recoveryEnabled);
eventPublisher.publish(event);
+ deleteMetaData.addDeletedHostComponent(componentName,
+ hostName,
+ getHost().getHostId(),
+ Long.toString(clusterId),
+ version);
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyDeleteFormer.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyDeleteFormer.java b/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyDeleteFormer.java
new file mode 100644
index 0000000..199f127
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyDeleteFormer.java
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.topology;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+import org.apache.ambari.server.agent.stomp.dto.TopologyComponent;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
+import org.apache.ambari.server.events.TopologyUpdateEvent;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+public class TopologyDeleteFormer {
+
+ @Inject
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
+ public void processDeleteMetaDataException(DeleteHostComponentStatusMetaData metaData) throws AmbariException {
+ if (metaData.getAmbariException() != null) {
+
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(
+ createUpdateFromDeleteMetaData(metaData),
+ TopologyUpdateEvent.EventType.DELETE
+ );
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
+
+ throw metaData.getAmbariException();
+ }
+ }
+ public void processDeleteMetaData(DeleteHostComponentStatusMetaData metaData) {
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(
+ createUpdateFromDeleteMetaData(metaData),
+ TopologyUpdateEvent.EventType.DELETE
+ );
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
+ }
+
+ public void processDeleteCluster(String clusterId) {
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
+ topologyUpdates.put(clusterId, new TopologyCluster());
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(
+ topologyUpdates,
+ TopologyUpdateEvent.EventType.DELETE
+ );
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
+ }
+
+ public Map<String, TopologyCluster> createUpdateFromDeleteMetaData(DeleteHostComponentStatusMetaData metaData) {
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
+
+ for (DeleteHostComponentStatusMetaData.HostComponent hostComponent : metaData.getRemovedHostComponents()) {
+ TopologyComponent deletedComponent = TopologyComponent.newBuilder()
+ .setComponentName(hostComponent.getComponentName())
+ .setVersion(hostComponent.getVersion())
+ .setHostIds(new HashSet<>(Arrays.asList(hostComponent.getHostId())))
+ .build();
+
+ String clusterId = hostComponent.getClusterId();
+ if (!topologyUpdates.containsKey(clusterId)) {
+ topologyUpdates.put(clusterId, new TopologyCluster());
+ }
+
+ if (!topologyUpdates.get(clusterId).getTopologyComponents().contains(deletedComponent)) {
+ topologyUpdates.get(clusterId).addTopologyComponent(deletedComponent);
+ } else {
+ topologyUpdates.get(clusterId).getTopologyComponents()
+ .stream().filter(t -> t.equals(deletedComponent))
+ .forEach(t -> t.addHostId(hostComponent.getHostId()));
+ }
+ }
+
+ return topologyUpdates;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java
index edf107a..eb030c6 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog200.java
@@ -480,6 +480,7 @@ public class UpgradeCatalog200 extends AbstractUpgradeCatalog {
return;
}
+ //TODO Event for hostcomponents removing
for (ClusterEntity cluster : clusters) {
ClusterServiceEntity nagios = clusterServiceDao.findByClusterAndServiceNames(
cluster.getClusterName(), "NAGIOS");
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java b/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
index 3a93fbf..e9a087a 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
@@ -36,6 +36,7 @@ import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.agent.HeartbeatTestHelper;
import org.apache.ambari.server.agent.RecoveryConfig;
import org.apache.ambari.server.agent.RecoveryConfigHelper;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
@@ -180,7 +181,7 @@ public class RecoveryConfigHelperTest {
assertEquals(recoveryConfig.getEnabledComponents(), "DATANODE,NAMENODE");
// Uninstall HDFS::DATANODE from host1
- hdfs.getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1).delete();
+ hdfs.getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1).delete(new DeleteHostComponentStatusMetaData());
// Verify that the config is stale
boolean isConfigStale = recoveryConfigHelper.isConfigStale(cluster.getClusterName(), DummyHostname1,
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java
index 7c34521..1f44279 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java
@@ -528,7 +528,7 @@ public class ComponentResourceProviderTest {
expect(serviceComponentHost.getDesiredState()).andReturn(hostComponentState);
- service.deleteServiceComponent("Component100");
+ service.deleteServiceComponent("Component100", new DeleteHostComponentStatusMetaData());
expectLastCall().once();
// replay
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java
index 44c2493..bad5854 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java
@@ -858,7 +858,7 @@ public class ServiceResourceProviderTest {
expect(service.getName()).andReturn(serviceName).anyTimes();
expect(service.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>());
expect(service.getCluster()).andReturn(cluster);
- cluster.deleteService(serviceName);
+ cluster.deleteService(serviceName, new DeleteHostComponentStatusMetaData());
// replay
replay(managementController, clusters, cluster, service);
@@ -906,7 +906,7 @@ public class ServiceResourceProviderTest {
expect(service.getName()).andReturn(serviceName).anyTimes();
expect(service.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>());
expect(service.getCluster()).andReturn(cluster);
- cluster.deleteService(serviceName);
+ cluster.deleteService(serviceName, new DeleteHostComponentStatusMetaData());
// replay
replay(managementController, clusters, cluster, service);
@@ -1074,7 +1074,7 @@ public class ServiceResourceProviderTest {
expect(sch3.canBeRemoved()).andReturn(sch3State.isRemovableState()).anyTimes();
expect(service.getCluster()).andReturn(cluster);
- cluster.deleteService(serviceName);
+ cluster.deleteService(serviceName, new DeleteHostComponentStatusMetaData());
// replay
replay(managementController, clusters, cluster, service,
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java b/ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java
index 6fb0028..b5fd75e 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import javax.persistence.EntityManager;
import org.apache.ambari.server.H2DatabaseCleaner;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
@@ -192,7 +193,7 @@ public class EventsTest {
AlertDefinitionEntity definition = hdfsDefinitions.get(0);
// delete HDFS
- m_cluster.getService("HDFS").delete();
+ m_cluster.getService("HDFS").delete(new DeleteHostComponentStatusMetaData());
// verify the event was received
Assert.assertTrue(m_listener.isAmbariEventReceived(eventClass));
@@ -242,7 +243,7 @@ public class EventsTest {
Assert.assertEquals(0, hdfsDefinitions.size());
// delete HDFS
- m_cluster.getService("HDFS").delete();
+ m_cluster.getService("HDFS").delete(new DeleteHostComponentStatusMetaData());
// verify the event was received
Assert.assertTrue(m_listener.isAmbariEventReceived(eventClass));
@@ -293,7 +294,7 @@ public class EventsTest {
Assert.assertNull(group);
// delete HDFS
- m_cluster.getService("HDFS").delete();
+ m_cluster.getService("HDFS").delete(new DeleteHostComponentStatusMetaData());
// verify the event was received
Assert.assertTrue(m_listener.isAmbariEventReceived(eventClass));
@@ -315,7 +316,7 @@ public class EventsTest {
installHdfsService();
Assert.assertFalse(m_listener.isAmbariEventReceived(eventClass));
- m_cluster.getServiceComponentHosts(HOSTNAME).get(0).delete();
+ m_cluster.getServiceComponentHosts(HOSTNAME).get(0).delete(new DeleteHostComponentStatusMetaData());
Assert.assertTrue(m_listener.isAmbariEventReceived(eventClass));
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListenerTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListenerTest.java b/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListenerTest.java
index 90b8687..896b576 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListenerTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/tasks/TaskStatusListenerTest.java
@@ -32,9 +32,11 @@ import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.events.TaskCreateEvent;
import org.apache.ambari.server.events.TaskUpdateEvent;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.events.publishers.TaskEventPublisher;
import org.apache.ambari.server.orm.dao.ExecutionCommandDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.RequestDAO;
import org.apache.ambari.server.orm.dao.StageDAO;
import org.apache.ambari.server.orm.entities.RequestEntity;
@@ -42,6 +44,7 @@ import org.apache.ambari.server.orm.entities.RoleSuccessCriteriaEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.orm.entities.StageEntityPK;
import org.apache.ambari.server.state.ServiceComponentHostEvent;
+import org.apache.ambari.server.topology.TopologyManager;
import org.easymock.EasyMock;
import org.easymock.EasyMockSupport;
import org.junit.Assert;
@@ -53,6 +56,7 @@ import com.google.inject.Inject;
public class TaskStatusListenerTest extends EasyMockSupport {
private TaskEventPublisher publisher = new TaskEventPublisher();
+ private StateUpdateEventPublisher statePublisher = new StateUpdateEventPublisher();
@Inject
private ExecutionCommandDAO executionCommandDAO;
@@ -85,6 +89,8 @@ public class TaskStatusListenerTest extends EasyMockSupport {
HostRoleStatus hostRoleStatus = HostRoleStatus.PENDING;
StageDAO stageDAO = createNiceMock(StageDAO.class);
+ HostRoleCommandDAO hostRoleCommandDAO = createNiceMock(HostRoleCommandDAO.class);
+ TopologyManager topologyManager = createNiceMock(TopologyManager.class);
RequestDAO requestDAO = createNiceMock(RequestDAO.class);
StageEntity stageEntity = createNiceMock(StageEntity.class);
RequestEntity requestEntity = createNiceMock(RequestEntity.class);
@@ -108,7 +114,8 @@ public class TaskStatusListenerTest extends EasyMockSupport {
EasyMock.replay(requestDAO);
TaskCreateEvent event = new TaskCreateEvent(hostRoleCommands);
- TaskStatusListener listener = new TaskStatusListener(publisher,stageDAO,requestDAO);
+ TaskStatusListener listener = new TaskStatusListener(publisher,stageDAO,requestDAO,statePublisher,
+ hostRoleCommandDAO,topologyManager);
Assert.assertTrue(listener.getActiveTasksMap().isEmpty());
Assert.assertTrue(listener.getActiveStageMap().isEmpty());
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java b/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java
index dc9ce5e..c0d445a 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java
@@ -34,6 +34,7 @@ import java.util.UUID;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.H2DatabaseCleaner;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
import org.apache.ambari.server.events.ServiceInstalledEvent;
@@ -502,7 +503,7 @@ public class HostVersionOutOfSyncListenerTest {
List<ServiceComponentHost> hostComponents = c1.getServiceComponentHosts(host3);
for (ServiceComponentHost sch : hostComponents) {
if (sch.getServiceName().equals("HDFS")) {
- sch.delete();
+ sch.delete(new DeleteHostComponentStatusMetaData());
StackId clusterStackId = c1.getDesiredStackVersion();
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceComponentTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceComponentTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceComponentTest.java
index 8e8bed3..711a3c6 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceComponentTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceComponentTest.java
@@ -33,6 +33,7 @@ import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ServiceComponentResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.OrmTestHelper;
@@ -438,17 +439,14 @@ public class ServiceComponentTest {
sch1.setState(State.STARTED);
sch2.setState(State.STARTED);
- try {
- // delete the SC
- sc.delete();
- Assert.assertTrue("Delete must fail as some SCH are in STARTED state", false);
- }catch(AmbariException e) {
- // expected
- }
+ // delete the SC
+ DeleteHostComponentStatusMetaData deleteMetaData = new DeleteHostComponentStatusMetaData();
+ sc.delete(deleteMetaData);
+ Assert.assertNull("Delete must fail as some SCH are in STARTED state", deleteMetaData.getAmbariException());
sch1.setState(State.INSTALLED);
sch2.setState(State.INSTALL_FAILED);
- sc.delete();
+ sc.delete(new DeleteHostComponentStatusMetaData());
// verify history is gone, too
serviceComponentDesiredStateEntity = serviceComponentDesiredStateDAO.findByName(
@@ -514,7 +512,7 @@ public class ServiceComponentTest {
assertEquals(1, componentHistoryList.size());
// delete the SC
- sc.delete();
+ sc.delete(new DeleteHostComponentStatusMetaData());
// verify history is gone, too
serviceComponentDesiredStateEntity = serviceComponentDesiredStateDAO.findByName(
@@ -628,7 +626,7 @@ public class ServiceComponentTest {
assertEquals(RepositoryVersionState.CURRENT, persistedVersion.getState());
- sc.delete();
+ sc.delete(new DeleteHostComponentStatusMetaData());
serviceComponentDesiredStateEntity = serviceComponentDesiredStateDAO.findByName(
cluster.getClusterId(), serviceName, componentName);
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceTest.java
index fc12e44..812ae36 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/ServiceTest.java
@@ -29,6 +29,7 @@ import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ServiceResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.dao.ClusterServiceDAO;
@@ -205,7 +206,7 @@ public class ServiceTest {
s.getServiceComponent("HDFS_CLIENT").getDesiredState());
// delete service component
- s.deleteServiceComponent("NAMENODE");
+ s.deleteServiceComponent("NAMENODE", new DeleteHostComponentStatusMetaData());
assertEquals(3, s.getServiceComponents().size());
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/state/alerts/AlertEventPublisherTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/alerts/AlertEventPublisherTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/alerts/AlertEventPublisherTest.java
index 8c23b69..5d8f6dd 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/alerts/AlertEventPublisherTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/alerts/AlertEventPublisherTest.java
@@ -20,6 +20,7 @@ package org.apache.ambari.server.state.alerts;
import java.util.UUID;
import org.apache.ambari.server.H2DatabaseCleaner;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.AlertDefinitionChangedEvent;
import org.apache.ambari.server.events.AlertDefinitionDeleteEvent;
import org.apache.ambari.server.events.AmbariEvent;
@@ -133,7 +134,7 @@ public class AlertEventPublisherTest {
Assert.assertEquals(0, dispatchDao.findAllGroups().size());
installHdfsService();
Assert.assertEquals(1, dispatchDao.findAllGroups().size());
- cluster.getService("HDFS").delete();
+ cluster.getService("HDFS").delete(new DeleteHostComponentStatusMetaData());
Assert.assertEquals(0, dispatchDao.findAllGroups().size());
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterImplTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterImplTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterImplTest.java
index 76f9130..3836d6b 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterImplTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterImplTest.java
@@ -38,6 +38,7 @@ import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.HostNotFoundException;
import org.apache.ambari.server.controller.AmbariSessionManager;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.state.Cluster;
@@ -242,7 +243,7 @@ public class ClusterImplTest {
ServiceComponentHost tezClientHost2 = tezClient.addServiceComponentHost(hostName2);
// When
- cluster.deleteService(serviceToDelete);
+ cluster.deleteService(serviceToDelete, new DeleteHostComponentStatusMetaData());
// Then
assertFalse("Deleted service should be removed from the service collection !", cluster.getServices().containsKey(serviceToDelete));
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java
index 76689f4..04c0ead 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java
@@ -58,6 +58,7 @@ import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.ClusterResponse;
import org.apache.ambari.server.controller.ConfigurationResponse;
import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
@@ -1109,7 +1110,7 @@ public class ClusterTest {
assertEquals(2, injector.getProvider(EntityManager.class).get().
createQuery("SELECT service FROM ClusterServiceEntity service").getResultList().size());
- c1.deleteService("HDFS");
+ c1.deleteService("HDFS", new DeleteHostComponentStatusMetaData());
assertEquals(1, c1.getServices().size());
assertEquals(1, injector.getProvider(EntityManager.class).get().
@@ -1146,7 +1147,7 @@ public class ClusterTest {
c1.getActiveServiceConfigVersions();
Assert.assertEquals(1, activeServiceConfigVersions.size());
- c1.deleteService("HDFS");
+ c1.deleteService("HDFS", new DeleteHostComponentStatusMetaData());
Assert.assertEquals(0, c1.getServices().size());
Assert.assertEquals(0, c1.getServiceConfigVersions().size());
[3/3] ambari git commit: AMBARI-21193. Add TopologyChange event to
the server (mpapirkovskyy)
Posted by mp...@apache.org.
AMBARI-21193. Add TopologyChange event to the server (mpapirkovskyy)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/c2ab4a3c
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/c2ab4a3c
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/c2ab4a3c
Branch: refs/heads/branch-3.0-perf
Commit: c2ab4a3ce19fba934f3766f30a4738d8ebbb061e
Parents: 5d09624
Author: Myroslav Papirkovskyi <mp...@hortonworks.com>
Authored: Wed Jun 7 18:50:54 2017 +0300
Committer: Myroslav Papirkovskyi <mp...@hortonworks.com>
Committed: Wed Jun 7 18:50:54 2017 +0300
----------------------------------------------------------------------
.../server/HostNotRegisteredException.java | 33 +++
.../actionmanager/ActionDBAccessorImpl.java | 10 +
.../server/agent/AgentSessionManager.java | 50 +++++
.../ambari/server/agent/HeartBeatHandler.java | 66 +++++-
.../ambari/server/agent/HeartbeatProcessor.java | 45 +++--
.../agent/stomp/AgentReportsController.java | 82 ++++++++
.../server/agent/stomp/HeartbeatController.java | 16 +-
.../agent/stomp/dto/ComponentStatusReport.java | 84 ++++++++
.../agent/stomp/dto/ComponentStatusReports.java | 45 +++++
.../server/agent/stomp/dto/TopologyCluster.java | 61 ++++++
.../agent/stomp/dto/TopologyComponent.java | 136 +++++++++++++
.../server/agent/stomp/dto/TopologyHost.java | 89 +++++++++
.../stomp/dto/TopologyStatusCommandParams.java | 47 +++++
.../server/api/services/AmbariMetaInfo.java | 25 +++
.../configuration/spring/ApiStompConfig.java | 9 +
.../configuration/spring/GuiceBeansConfig.java | 2 -
.../AmbariManagementControllerImpl.java | 73 +++++--
.../controller/internal/CalculatedStatus.java | 33 +++
.../internal/ComponentResourceProvider.java | 24 ++-
.../DeleteHostComponentStatusMetaData.java | 98 +++++++++
.../internal/HostResourceProvider.java | 62 ++++++
.../internal/RequestResourceProvider.java | 24 +--
.../internal/ServiceResourceProvider.java | 9 +-
.../ambari/server/events/AlertUpdateEvent.java | 45 +++++
.../ambari/server/events/AmbariUpdateEvent.java | 26 +++
.../ambari/server/events/CommandEvent.java | 25 +++
.../server/events/ConfigsUpdateEvent.java | 200 +++++++++++++++++++
.../server/events/HostComponentUpdateEvent.java | 94 +++++++++
.../server/events/MetadataUpdateEvent.java | 25 +++
.../server/events/RequestUpdateEvent.java | 108 ++++++++++
.../server/events/TopologyUpdateEvent.java | 63 ++++++
.../listeners/alerts/AlertReceivedListener.java | 10 +
.../listeners/requests/StateUpdateListener.java | 52 +++++
.../listeners/tasks/TaskStatusListener.java | 23 ++-
.../publishers/StateUpdateEventPublisher.java | 45 +++++
.../ambari/server/orm/dao/RequestDAO.java | 4 +-
.../server/orm/entities/ResourceEntity.java | 2 -
.../org/apache/ambari/server/state/Cluster.java | 3 +-
.../org/apache/ambari/server/state/Service.java | 7 +-
.../ambari/server/state/ServiceComponent.java | 7 +-
.../server/state/ServiceComponentHost.java | 3 +-
.../server/state/ServiceComponentImpl.java | 21 +-
.../apache/ambari/server/state/ServiceImpl.java | 33 ++-
.../server/state/cluster/ClusterImpl.java | 52 ++++-
.../svccomphost/ServiceComponentHostImpl.java | 33 ++-
.../server/topology/TopologyDeleteFormer.java | 98 +++++++++
.../server/upgrade/UpgradeCatalog200.java | 1 +
.../configuration/RecoveryConfigHelperTest.java | 3 +-
.../internal/ComponentResourceProviderTest.java | 2 +-
.../internal/ServiceResourceProviderTest.java | 6 +-
.../apache/ambari/server/events/EventsTest.java | 9 +-
.../listeners/tasks/TaskStatusListenerTest.java | 9 +-
.../HostVersionOutOfSyncListenerTest.java | 3 +-
.../server/state/ServiceComponentTest.java | 18 +-
.../apache/ambari/server/state/ServiceTest.java | 3 +-
.../state/alerts/AlertEventPublisherTest.java | 3 +-
.../server/state/cluster/ClusterImplTest.java | 3 +-
.../server/state/cluster/ClusterTest.java | 5 +-
.../server/testing/DBInconsistencyTests.java | 3 +-
59 files changed, 2036 insertions(+), 134 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/HostNotRegisteredException.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/HostNotRegisteredException.java b/ambari-server/src/main/java/org/apache/ambari/server/HostNotRegisteredException.java
new file mode 100644
index 0000000..9e7b3a7
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/HostNotRegisteredException.java
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server;
+
+/**
+ * Thrown when an attempt is made to retrieve host not registered in cluster.
+ */
+public class HostNotRegisteredException extends AmbariException {
+
+ /**
+ * Constructor
+ * @param sessionId sessionId of websocket message
+ */
+ public HostNotRegisteredException(String sessionId) {
+ super(String.format("Host with [%s] sessionId not registered", sessionId));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java
index 8c4eae8..d372844 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java
@@ -45,8 +45,10 @@ import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.internal.CalculatedStatus;
import org.apache.ambari.server.events.HostsRemovedEvent;
import org.apache.ambari.server.events.RequestFinishedEvent;
+import org.apache.ambari.server.events.RequestUpdateEvent;
import org.apache.ambari.server.events.TaskCreateEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.events.publishers.TaskEventPublisher;
import org.apache.ambari.server.orm.dao.ClusterDAO;
import org.apache.ambari.server.orm.dao.ExecutionCommandDAO;
@@ -67,6 +69,7 @@ import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.security.authorization.AuthorizationHelper;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Host;
+import org.apache.ambari.server.topology.TopologyManager;
import org.apache.ambari.server.utils.LoopBody;
import org.apache.ambari.server.utils.Parallel;
import org.apache.ambari.server.utils.ParallelLoopResult;
@@ -137,6 +140,12 @@ public class ActionDBAccessorImpl implements ActionDBAccessor {
@Inject
AuditLogger auditLogger;
+ @Inject
+ StateUpdateEventPublisher stateUpdateEventPublisher;
+
+ @Inject
+ TopologyManager topologyManager;
+
/**
* Cache for auditlog. It stores a {@link RequestDetails} object for every requests.
* {@link RequestDetails} contains the previous status of the request and a map for tasks.
@@ -360,6 +369,7 @@ public class ActionDBAccessorImpl implements ActionDBAccessor {
requestEntity.setClusterId(clusterId);
requestDAO.create(requestEntity);
+ stateUpdateEventPublisher.publish(new RequestUpdateEvent(requestEntity, hostRoleCommandDAO, topologyManager));
//TODO wire request to cluster
List<StageEntity> stageEntities = new ArrayList<>(request.getStages().size());
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/AgentSessionManager.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/AgentSessionManager.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/AgentSessionManager.java
new file mode 100644
index 0000000..d9b595a
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/AgentSessionManager.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.ambari.server.HostNotRegisteredException;
+import org.apache.ambari.server.state.Host;
+
+import com.google.inject.Singleton;
+
+@Singleton
+public class AgentSessionManager {
+ private static ConcurrentHashMap<String, Host> registeredHosts = new ConcurrentHashMap<>();
+
+ public void register(String sessionId, Host registeredHost) {
+ String existKey = registeredHosts.entrySet().stream()
+ .filter(e -> e.getValue().getHostName().equals(registeredHost.getHostName())).map(Map.Entry::getKey)
+ .findAny().orElse(null);
+ if (existKey != null) {
+ registeredHosts.remove(existKey);
+ }
+ registeredHosts.put(sessionId, registeredHost);
+ }
+
+ public Host getHost(String sessionId) throws HostNotRegisteredException {
+ if (registeredHosts.containsKey(sessionId)) {
+ return registeredHosts.get(sessionId);
+ }
+ throw new HostNotRegisteredException(sessionId);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
index 3601528..67ec87d 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
@@ -22,18 +22,25 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.HostNotFoundException;
import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+import org.apache.ambari.server.agent.stomp.dto.TopologyComponent;
+import org.apache.ambari.server.agent.stomp.dto.TopologyHost;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.events.TopologyUpdateEvent;
import org.apache.ambari.server.serveraction.kerberos.KerberosIdentityDataFileReader;
import org.apache.ambari.server.serveraction.kerberos.KerberosIdentityDataFileReaderFactory;
import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
@@ -44,6 +51,8 @@ import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostState;
+import org.apache.ambari.server.state.Service;
+import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
@@ -260,7 +269,9 @@ public class HeartBeatHandler {
return response;
}
-
+ public void handleComponentReportStatus(List<ComponentStatus> componentStatuses, String hostname) throws AmbariException {
+ heartbeatProcessor.processStatusReports(componentStatuses, hostname);
+ }
protected void processRecoveryReport(RecoveryReport recoveryReport, String hostname) throws AmbariException {
LOG.debug("Received recovery report: " + recoveryReport.toString());
@@ -470,6 +481,59 @@ public class HeartBeatHandler {
}
/**
+ * Is used during agent registering to provide base info about clusters topology.
+ * @return filled TopologyUpdateEvent with info about all components and hosts in all clusters
+ * @throws InvalidStateTransitionException
+ * @throws AmbariException
+ */
+ //TODO need move to better place
+ public TopologyUpdateEvent getInitialClusterTopology()
+ throws InvalidStateTransitionException, AmbariException {
+ Map<String, TopologyCluster> topologyClusters = new HashMap<>();
+ for (Cluster cl : clusterFsm.getClusters().values()) {
+ Collection<Host> clusterHosts = cl.getHosts();
+ Set<TopologyComponent> topologyComponents = new HashSet<>();
+ Set<TopologyHost> topologyHosts = new HashSet<>();
+ for (Host host : clusterHosts) {
+ topologyHosts.add(new TopologyHost(host.getHostId(), host.getHostName(),
+ host.getRackInfo(), host.getIPv4()));
+ }
+ for (Service service : cl.getServices().values()) {
+ for (ServiceComponent component : service.getServiceComponents().values()) {
+ Map<String, ServiceComponentHost> componentsMap = component.getServiceComponentHosts();
+ if (!componentsMap.isEmpty()) {
+
+ //TODO will be a need to change to multi-instance usage
+ ServiceComponentHost sch = componentsMap.entrySet().iterator().next().getValue();
+
+ Set<String> hostNames = cl.getHosts(sch.getServiceName(), sch.getServiceComponentName());
+ Set<Long> hostOrderIds = clusterHosts.stream()
+ .filter(h -> hostNames.contains(h.getHostName()))
+ .map(h -> h.getHostId()).collect(Collectors.toSet());
+ String serviceName = sch.getServiceName();
+ String componentName = sch.getServiceComponentName();
+ StackId stackId = cl.getDesiredStackVersion();
+
+ TopologyComponent topologyComponent = TopologyComponent.newBuilder()
+ .setComponentName(sch.getServiceComponentName())
+ .setServiceName(sch.getServiceName())
+ .setVersion(sch.getVersion())
+ .setHostIds(hostOrderIds)
+ .setStatusCommandParams(ambariMetaInfo.getStatusCommandParams(stackId, serviceName, componentName))
+ .build();
+ topologyComponents.add(topologyComponent);
+ }
+ }
+ }
+ topologyClusters.put(Long.toString(cl.getClusterId()),
+ new TopologyCluster(topologyComponents, topologyHosts));
+ }
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(topologyClusters,
+ TopologyUpdateEvent.EventType.UPDATE);
+ return topologyUpdateEvent;
+ }
+
+ /**
* Annotate the response with some housekeeping details.
* hasMappedComponents - indicates if any components are mapped to the host
* hasPendingTasks - indicates if any tasks are pending for the host (they may not be sent yet)
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
index c1028dc..47109df 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
@@ -583,10 +583,17 @@ public class HeartbeatProcessor extends AbstractService{
* @throws AmbariException
*/
protected void processStatusReports(HeartBeat heartbeat) throws AmbariException {
- String hostname = heartbeat.getHostname();
+ processStatusReports(heartbeat.getComponentStatus(), heartbeat.getHostname());
+ }
+
+ /**
+ * Process reports of status commands
+ * @throws AmbariException
+ */
+ public void processStatusReports(List<ComponentStatus> componentStatuses, String hostname) throws AmbariException {
Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);
for (Cluster cl : clusters) {
- for (ComponentStatus status : heartbeat.componentStatus) {
+ for (ComponentStatus status : componentStatuses) {
if (status.getClusterName().equals(cl.getClusterName())) {
try {
Service svc = cl.getService(status.getServiceName());
@@ -597,22 +604,24 @@ public class HeartbeatProcessor extends AbstractService{
componentName);
ServiceComponentHost scHost = svcComp.getServiceComponentHost(
hostname);
- org.apache.ambari.server.state.State prevState = scHost.getState();
- org.apache.ambari.server.state.State liveState =
- org.apache.ambari.server.state.State.valueOf(org.apache.ambari.server.state.State.class,
- status.getStatus());
- //ignore reports from status commands if component is in INIT or any "in progress" state
- if (prevState.equals(org.apache.ambari.server.state.State.INSTALLED)
- || prevState.equals(org.apache.ambari.server.state.State.STARTED)
- || prevState.equals(org.apache.ambari.server.state.State.UNKNOWN)) {
- scHost.setState(liveState);
- if (!prevState.equals(liveState)) {
- LOG.info("State of service component " + componentName
- + " of service " + status.getServiceName()
- + " of cluster " + status.getClusterName()
- + " has changed from " + prevState + " to " + liveState
- + " at host " + hostname
- + " according to STATUS_COMMAND report");
+ if (status.getStatus() != null) {
+ org.apache.ambari.server.state.State prevState = scHost.getState();
+ org.apache.ambari.server.state.State liveState =
+ org.apache.ambari.server.state.State.valueOf(org.apache.ambari.server.state.State.class,
+ status.getStatus());
+ //ignore reports from status commands if component is in INIT or any "in progress" state
+ if (prevState.equals(org.apache.ambari.server.state.State.INSTALLED)
+ || prevState.equals(org.apache.ambari.server.state.State.STARTED)
+ || prevState.equals(org.apache.ambari.server.state.State.UNKNOWN)) {
+ scHost.setState(liveState);
+ if (!prevState.equals(liveState)) {
+ LOG.info("State of service component " + componentName
+ + " of service " + status.getServiceName()
+ + " of cluster " + status.getClusterName()
+ + " has changed from " + prevState + " to " + liveState
+ + " at host " + hostname
+ + " according to STATUS_COMMAND report");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
new file mode 100644
index 0000000..68b7f3b
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent.stomp;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import javax.ws.rs.WebApplicationException;
+
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.agent.AgentSessionManager;
+import org.apache.ambari.server.agent.ComponentStatus;
+import org.apache.ambari.server.agent.HeartBeatHandler;
+import org.apache.ambari.server.agent.stomp.dto.ComponentStatusReport;
+import org.apache.ambari.server.agent.stomp.dto.ComponentStatusReports;
+import org.apache.ambari.server.state.cluster.ClustersImpl;
+import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.simp.annotation.SendToUser;
+import org.springframework.messaging.simp.annotation.SubscribeMapping;
+import org.springframework.stereotype.Controller;
+
+import com.google.inject.Injector;
+
+@Controller
+@SendToUser("/")
+@MessageMapping("/reports")
+public class AgentReportsController {
+ private static Log LOG = LogFactory.getLog(AgentReportsController.class);
+ private final HeartBeatHandler hh;
+ private final ClustersImpl clusters;
+ private final AgentSessionManager agentSessionManager;
+
+ public AgentReportsController(Injector injector) {
+ hh = injector.getInstance(HeartBeatHandler.class);
+ clusters = injector.getInstance(ClustersImpl.class);
+ agentSessionManager = injector.getInstance(AgentSessionManager.class);
+ }
+
+ @SubscribeMapping("/component_status")
+ public void handleComponentReportStatus(@Header String simpSessionId, ComponentStatusReports message)
+ throws WebApplicationException, InvalidStateTransitionException, AmbariException {
+ List<ComponentStatus> statuses = new ArrayList<>();
+ for (Map.Entry<String, List<ComponentStatusReport>> clusterReport : message.getComponentStatusReports().entrySet()) {
+ for (ComponentStatusReport report : clusterReport.getValue()) {
+ ComponentStatus componentStatus = new ComponentStatus();
+ componentStatus.setClusterName(clusters.getCluster(report.getClusterId()).getClusterName());
+ componentStatus.setComponentName(report.getComponentName());
+ componentStatus.setServiceName(report.getServiceName());
+ if (report.getCommand().equals(ComponentStatusReport.CommandStatusCommand.STATUS)) {
+ componentStatus.setStatus(report.getStatus().toString());
+ } else {
+ componentStatus.setSecurityState(report.getStatus().toString());
+ }
+ statuses.add(componentStatus);
+ }
+ }
+
+ hh.handleComponentReportStatus(statuses,
+ agentSessionManager.getHost(simpSessionId).getHostName());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
index 510f200..b56449e 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
@@ -20,15 +20,19 @@ package org.apache.ambari.server.agent.stomp;
import javax.ws.rs.WebApplicationException;
import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.agent.AgentSessionManager;
import org.apache.ambari.server.agent.HeartBeat;
import org.apache.ambari.server.agent.HeartBeatHandler;
import org.apache.ambari.server.agent.HeartBeatResponse;
import org.apache.ambari.server.agent.Register;
import org.apache.ambari.server.agent.RegistrationResponse;
import org.apache.ambari.server.agent.RegistrationStatus;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
+import org.apache.ambari.server.state.cluster.ClustersImpl;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.annotation.SendToUser;
import org.springframework.messaging.simp.annotation.SubscribeMapping;
@@ -42,14 +46,21 @@ import com.google.inject.Injector;
public class HeartbeatController {
private static Log LOG = LogFactory.getLog(HeartbeatController.class);
private final HeartBeatHandler hh;
+ private final ClustersImpl clusters;
+ private final AgentSessionManager agentSessionManager;
+ private final StateUpdateEventPublisher stateUpdateEventPublisher;
public HeartbeatController(Injector injector) {
hh = injector.getInstance(HeartBeatHandler.class);
+ clusters = injector.getInstance(ClustersImpl.class);
+ agentSessionManager = injector.getInstance(AgentSessionManager.class);
+ stateUpdateEventPublisher = injector.getInstance(StateUpdateEventPublisher.class);
}
@SubscribeMapping("/register")
- public RegistrationResponse register(Register message)
- throws WebApplicationException, InvalidStateTransitionException {
+ public RegistrationResponse register(@Header String simpSessionId, Register message)
+ throws WebApplicationException, InvalidStateTransitionException, AmbariException {
+
/* Call into the heartbeat handler */
RegistrationResponse response = null;
@@ -64,6 +75,7 @@ public class HeartbeatController {
response.setLog(ex.getMessage());
return response;
}
+ stateUpdateEventPublisher.publish(hh.getInitialClusterTopology());
return response;
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReport.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReport.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReport.java
new file mode 100644
index 0000000..b430510
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReport.java
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.agent.stomp.dto;
+
+public class ComponentStatusReport {
+ private String componentName;
+ private CommandStatusCommand command;
+ private String status;
+ private String serviceName;
+ private Long clusterId;
+
+ public ComponentStatusReport() {
+ }
+
+ public ComponentStatusReport(String componentName, CommandStatusCommand command, String status, String serviceName, Long clusterId) {
+ this.componentName = componentName;
+ this.command = command;
+ this.status = status;
+ this.serviceName = serviceName;
+ this.clusterId = clusterId;
+ }
+
+ public String getComponentName() {
+ return componentName;
+ }
+
+ public void setComponentName(String componentName) {
+ this.componentName = componentName;
+ }
+
+ public CommandStatusCommand getCommand() {
+ return command;
+ }
+
+ public void setCommand(CommandStatusCommand command) {
+ this.command = command;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ public Long getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(Long clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public enum CommandStatusCommand {
+ STATUS,
+ SECURITY_STATUS
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
new file mode 100644
index 0000000..2a3e4bc
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.agent.stomp.dto;
+
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class ComponentStatusReports {
+
+ @JsonProperty("clustersReports")
+ private Map<String, List<ComponentStatusReport>> componentStatusReports;
+
+ public ComponentStatusReports() {
+ }
+
+ public ComponentStatusReports(Map<String, List<ComponentStatusReport>> componentStatusReports) {
+ this.componentStatusReports = componentStatusReports;
+ }
+
+ public Map<String, List<ComponentStatusReport>> getComponentStatusReports() {
+ return componentStatusReports;
+ }
+
+ public void setComponentStatusReports(Map<String, List<ComponentStatusReport>> componentStatusReports) {
+ this.componentStatusReports = componentStatusReports;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyCluster.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyCluster.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyCluster.java
new file mode 100644
index 0000000..4ddb8a6
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyCluster.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent.stomp.dto;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+public class TopologyCluster {
+ private Set<TopologyComponent> topologyComponents = new HashSet<>();
+ private Set<TopologyHost> topologyHosts = new HashSet<>();
+
+ public TopologyCluster() {
+ }
+
+ public TopologyCluster(Set<TopologyComponent> topologyComponents, Set<TopologyHost> topologyHosts) {
+ this.topologyComponents = topologyComponents;
+ this.topologyHosts = topologyHosts;
+ }
+
+ public Set<TopologyComponent> getTopologyComponents() {
+ return topologyComponents;
+ }
+
+ public void setTopologyComponents(Set<TopologyComponent> topologyComponents) {
+ this.topologyComponents = topologyComponents;
+ }
+
+ public Set<TopologyHost> getTopologyHosts() {
+ return topologyHosts;
+ }
+
+ public void setTopologyHosts(Set<TopologyHost> topologyHosts) {
+ this.topologyHosts = topologyHosts;
+ }
+
+ public void addTopologyHost(TopologyHost topologyHost) {
+ topologyHosts.add(topologyHost);
+ }
+
+ public void addTopologyComponent(TopologyComponent topologyComponent) {
+ topologyComponents.add(topologyComponent);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyComponent.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyComponent.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyComponent.java
new file mode 100644
index 0000000..019c933
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyComponent.java
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.agent.stomp.dto;
+
+import java.util.Set;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class TopologyComponent {
+ private String componentName;
+ private String serviceName;
+ private String version;
+ private Set<Long> hostIds;
+ private TopologyStatusCommandParams statusCommandParams;
+
+ private TopologyComponent() {
+ }
+
+ public static Builder newBuilder() {
+ return new TopologyComponent().new Builder();
+ }
+
+ public class Builder {
+ private Builder() {
+
+ }
+
+ public Builder setComponentName(String componentName) {
+ TopologyComponent.this.setComponentName(componentName);
+ return this;
+ }
+
+ public Builder setServiceName(String serviceName) {
+ TopologyComponent.this.setServiceName(serviceName);
+ return this;
+ }
+
+ public Builder setVersion(String version) {
+ TopologyComponent.this.setVersion(version);
+ return this;
+ }
+
+ public Builder setHostIds(Set<Long> hostIds) {
+ TopologyComponent.this.setHostIds(hostIds);
+ return this;
+ }
+
+ public Builder setStatusCommandParams(TopologyStatusCommandParams statusCommandParams) {
+ TopologyComponent.this.setStatusCommandParams(statusCommandParams);
+ return this;
+ }
+
+ public TopologyComponent build() {
+ return TopologyComponent.this;
+ }
+ }
+
+ public String getComponentName() {
+ return componentName;
+ }
+
+ public void setComponentName(String componentName) {
+ this.componentName = componentName;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ public void setVersion(String version) {
+ this.version = version;
+ }
+
+ public Set<Long> getHostIds() {
+ return hostIds;
+ }
+
+ public void setHostIds(Set<Long> hostIds) {
+ this.hostIds = hostIds;
+ }
+
+ public void addHostId(Long hostId) {
+ this.hostIds.add(hostId);
+ }
+
+ public TopologyStatusCommandParams getStatusCommandParams() {
+ return statusCommandParams;
+ }
+
+ public void setStatusCommandParams(TopologyStatusCommandParams statusCommandParams) {
+ this.statusCommandParams = statusCommandParams;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TopologyComponent that = (TopologyComponent) o;
+
+ if (!componentName.equals(that.componentName)) return false;
+ return version.equals(that.version);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = componentName.hashCode();
+ result = 31 * result + version.hashCode();
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyHost.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyHost.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyHost.java
new file mode 100644
index 0000000..1718711
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyHost.java
@@ -0,0 +1,89 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent.stomp.dto;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class TopologyHost {
+ private Long hostId;
+ private String hostName;
+ private String rackName;
+ private String ipv4;
+
+ public TopologyHost() {
+ }
+
+ public TopologyHost(Long hostId) {
+ this.hostId = hostId;
+ }
+
+ public TopologyHost(Long hostId, String hostName, String rackName, String ipv4) {
+ this.hostId = hostId;
+ this.hostName = hostName;
+ this.rackName = rackName;
+ this.ipv4 = ipv4;
+ }
+
+ public Long getHostId() {
+ return hostId;
+ }
+
+ public void setHostId(Long hostId) {
+ this.hostId = hostId;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ public void setHostName(String hostName) {
+ this.hostName = hostName;
+ }
+
+ public String getRackName() {
+ return rackName;
+ }
+
+ public void setRackName(String rackName) {
+ this.rackName = rackName;
+ }
+
+ public String getIpv4() {
+ return ipv4;
+ }
+
+ public void setIpv4(String ipv4) {
+ this.ipv4 = ipv4;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TopologyHost that = (TopologyHost) o;
+
+ return hostId.equals(that.hostId);
+ }
+
+ @Override
+ public int hashCode() {
+ return hostId.hashCode();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyStatusCommandParams.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyStatusCommandParams.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyStatusCommandParams.java
new file mode 100644
index 0000000..1574066
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/TopologyStatusCommandParams.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent.stomp.dto;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class TopologyStatusCommandParams {
+ private String script;
+ private String servicePackageFolder;
+
+ public TopologyStatusCommandParams(String script, String servicePackageFolder) {
+ this.script = script;
+ this.servicePackageFolder = servicePackageFolder;
+ }
+
+ public String getScript() {
+ return script;
+ }
+
+ public void setScript(String script) {
+ this.script = script;
+ }
+
+ public String getServicePackageFolder() {
+ return servicePackageFolder;
+ }
+
+ public void setServicePackageFolder(String servicePackageFolder) {
+ this.servicePackageFolder = servicePackageFolder;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java b/ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java
index c655c62..4887d52 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java
@@ -43,6 +43,7 @@ import javax.xml.bind.JAXBException;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.ParentObjectNotFoundException;
import org.apache.ambari.server.StackAccessException;
+import org.apache.ambari.server.agent.stomp.dto.TopologyStatusCommandParams;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.RootServiceResponseFactory.Services;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
@@ -61,6 +62,7 @@ import org.apache.ambari.server.stack.StackManager;
import org.apache.ambari.server.stack.StackManagerFactory;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.CommandScriptDefinition;
import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.DependencyInfo;
import org.apache.ambari.server.state.ExtensionInfo;
@@ -1486,5 +1488,28 @@ public class AmbariMetaInfo {
return versionDefinitions;
}
+ //TODO will be a need to change to multi-instance usage
+ public TopologyStatusCommandParams getStatusCommandParams(StackId stackId, String serviceName, String componentName) throws AmbariException {
+ ServiceInfo serviceInfo = getService(stackId.getStackName(),
+ stackId.getStackVersion(), serviceName);
+ ComponentInfo componentInfo = getComponent(
+ stackId.getStackName(), stackId.getStackVersion(),
+ serviceName, componentName);
+
+ String scriptName = null;
+ CommandScriptDefinition script = componentInfo.getCommandScript();
+ if (serviceInfo.getSchemaVersion().equals(AmbariMetaInfo.SCHEMA_VERSION_2)) {
+ if (script != null) {
+ scriptName = script.getScript();
+ } else {
+ String message = String.format("Component %s of service %s has not " +
+ "command script defined", componentName, serviceName);
+ throw new AmbariException(message);
+ }
+ }
+ String servicePackageFolder = serviceInfo.getServicePackageFolder();
+ return new TopologyStatusCommandParams(scriptName, servicePackageFolder);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/ApiStompConfig.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/ApiStompConfig.java b/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/ApiStompConfig.java
index f248e17..0d8a08b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/ApiStompConfig.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/ApiStompConfig.java
@@ -18,8 +18,10 @@
package org.apache.ambari.server.configuration.spring;
import org.apache.ambari.server.api.stomp.TestController;
+import org.apache.ambari.server.events.listeners.requests.StateUpdateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@@ -27,6 +29,8 @@ import org.springframework.web.socket.config.annotation.AbstractWebSocketMessage
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import com.google.inject.Injector;
+
@Configuration
@EnableWebSocketMessageBroker
@ComponentScan(basePackageClasses = {TestController.class})
@@ -34,6 +38,11 @@ import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
public class ApiStompConfig extends AbstractWebSocketMessageBrokerConfigurer {
private static final Logger LOG = LoggerFactory.getLogger(ApiStompConfig.class);
+ @Bean
+ public StateUpdateListener requestStatusListener(Injector injector) {
+ return new StateUpdateListener(injector);
+ }
+
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/v1")
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/GuiceBeansConfig.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/GuiceBeansConfig.java b/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/GuiceBeansConfig.java
index 52e5a98..50c3aba 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/GuiceBeansConfig.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/configuration/spring/GuiceBeansConfig.java
@@ -23,13 +23,11 @@ import org.apache.ambari.server.security.authorization.AmbariLocalUserProvider;
import org.apache.ambari.server.security.authorization.AmbariPamAuthenticationProvider;
import org.apache.ambari.server.security.authorization.AmbariUserAuthorizationFilter;
import org.apache.ambari.server.security.authorization.PermissionHelper;
-import org.apache.ambari.server.security.authorization.Users;
import org.apache.ambari.server.security.authorization.internal.AmbariInternalAuthenticationProvider;
import org.apache.ambari.server.security.ldap.AmbariLdapDataPopulator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
-import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
import org.springframework.security.crypto.password.PasswordEncoder;
import com.google.inject.Injector;
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
index 751ce08..52497f2 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
@@ -65,6 +65,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import javax.persistence.RollbackException;
@@ -88,10 +89,13 @@ import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.actionmanager.StageFactory;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+import org.apache.ambari.server.agent.stomp.dto.TopologyComponent;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.api.services.LoggingService;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.configuration.Configuration.DatabaseType;
+import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.controller.internal.DeleteStatusMetaData;
import org.apache.ambari.server.controller.internal.RequestOperationLevel;
import org.apache.ambari.server.controller.internal.RequestResourceFilter;
@@ -105,7 +109,9 @@ import org.apache.ambari.server.controller.metrics.MetricsCollectorHAManager;
import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.customactions.ActionDefinition;
+import org.apache.ambari.server.events.TopologyUpdateEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.metadata.ActionMetadata;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.metadata.RoleCommandOrderProvider;
@@ -200,6 +206,7 @@ import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartEvent
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStopEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostUpgradeEvent;
import org.apache.ambari.server.topology.Setting;
+import org.apache.ambari.server.topology.TopologyDeleteFormer;
import org.apache.ambari.server.utils.SecretReference;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.collections.CollectionUtils;
@@ -316,6 +323,12 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
@Inject
private StackDAO stackDAO;
+ @Inject
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
+ @Inject
+ TopologyDeleteFormer topologyDeleteFormer;
+
/**
* The KerberosHelper to help setup for enabling for disabling Kerberos
*/
@@ -676,6 +689,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
setMonitoringServicesRestartRequired(requests);
// now doing actual work
persistServiceComponentHosts(requests);
+ stateUpdateEventPublisher.publish(getAddedComponentsTopologyEvent(requests));
}
void persistServiceComponentHosts(Set<ServiceComponentHostRequest> requests)
@@ -707,6 +721,40 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
}
}
+ private TopologyUpdateEvent getAddedComponentsTopologyEvent(Set<ServiceComponentHostRequest> requests)
+ throws AmbariException {
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
+ for (ServiceComponentHostRequest request : requests) {
+ String serviceName = request.getServiceName();
+ String componentName = request.getComponentName();
+ Cluster cluster = clusters.getCluster(request.getClusterName());
+ Collection<Host> clusterHosts = cluster.getHosts();
+ Service s = cluster.getService(serviceName);
+ ServiceComponent sc = s.getServiceComponent(componentName);
+ Set<String> hostNames = cluster.getHosts(serviceName, componentName);
+ Set<Long> hostIds = clusterHosts.stream()
+ .filter(h -> hostNames.contains(h.getHostName()))
+ .map(h -> h.getHostId()).collect(Collectors.toSet());
+ ServiceComponentHost sch = sc.getServiceComponentHost(request.getHostname());
+
+ StackId stackId = cluster.getDesiredStackVersion();
+
+ TopologyComponent newComponent = TopologyComponent.newBuilder()
+ .setComponentName(sch.getServiceComponentName())
+ .setServiceName(sch.getServiceName())
+ .setVersion(sch.getVersion())
+ .setHostIds(hostIds)
+ .setStatusCommandParams(ambariMetaInfo.getStatusCommandParams(stackId, serviceName, componentName))
+ .build();
+ String clusterId = Long.toString(cluster.getClusterId());
+ if (!topologyUpdates.containsKey(clusterId)) {
+ topologyUpdates.put(clusterId, new TopologyCluster());
+ }
+ topologyUpdates.get(clusterId).addTopologyComponent(newComponent);
+ }
+ return new TopologyUpdateEvent(topologyUpdates, TopologyUpdateEvent.EventType.ADD);
+ }
+
private void setMonitoringServicesRestartRequired(
Set<ServiceComponentHostRequest> requests) throws AmbariException {
@@ -3434,7 +3482,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
}
Map<ServiceComponent, Set<ServiceComponentHost>> safeToRemoveSCHs = new HashMap<>();
- DeleteStatusMetaData deleteStatusMetaData = new DeleteStatusMetaData();
+ DeleteHostComponentStatusMetaData deleteMetaData = new DeleteHostComponentStatusMetaData();
for (ServiceComponentHostRequest request : expanded) {
@@ -3465,29 +3513,28 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
}
safeToRemoveSCHs.get(component).add(componentHost);
} catch (Exception ex) {
- deleteStatusMetaData.addException(request.getHostname() + "/" + request.getComponentName(), ex);
+ deleteMetaData.addException(request.getHostname() + "/" + request.getComponentName(), ex);
}
}
for (Entry<ServiceComponent, Set<ServiceComponentHost>> entry : safeToRemoveSCHs.entrySet()) {
for (ServiceComponentHost componentHost : entry.getValue()) {
try {
- deleteHostComponent(entry.getKey(), componentHost);
- deleteStatusMetaData.addDeletedKey(componentHost.getHostName() + "/" + componentHost.getServiceComponentName());
-
+ deleteHostComponent(entry.getKey(), componentHost, deleteMetaData);
} catch (Exception ex) {
- deleteStatusMetaData.addException(componentHost.getHostName() + "/" + componentHost.getServiceComponentName(), ex);
+ deleteMetaData.addException(componentHost.getHostName() + "/" + componentHost.getServiceComponentName(), ex);
}
}
}
//Do not break behavior for existing clients where delete request contains only 1 host component.
//Response for these requests will have empty body with appropriate error code.
- if (deleteStatusMetaData.getDeletedKeys().size() + deleteStatusMetaData.getExceptionForKeys().size() == 1) {
- if (deleteStatusMetaData.getDeletedKeys().size() == 1) {
+ if (deleteMetaData.getDeletedKeys().size() + deleteMetaData.getExceptionForKeys().size() == 1) {
+ if (deleteMetaData.getDeletedKeys().size() == 1) {
+ topologyDeleteFormer.processDeleteMetaData(deleteMetaData);
return null;
}
- Exception ex = deleteStatusMetaData.getExceptionForKeys().values().iterator().next();
+ Exception ex = deleteMetaData.getExceptionForKeys().values().iterator().next();
if (ex instanceof AmbariException) {
throw (AmbariException)ex;
} else {
@@ -3499,10 +3546,12 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
if (!safeToRemoveSCHs.isEmpty()) {
setMonitoringServicesRestartRequired(requests);
}
- return deleteStatusMetaData;
+ topologyDeleteFormer.processDeleteMetaData(deleteMetaData);
+ return deleteMetaData;
}
- private void deleteHostComponent(ServiceComponent serviceComponent, ServiceComponentHost componentHost) throws AmbariException {
+ private void deleteHostComponent(ServiceComponent serviceComponent, ServiceComponentHost componentHost,
+ DeleteHostComponentStatusMetaData deleteMetaData) throws AmbariException {
String included_hostname = componentHost.getHostName();
String serviceName = serviceComponent.getServiceName();
String master_component_name = null;
@@ -3510,7 +3559,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
HostComponentAdminState desiredAdminState = componentHost.getComponentAdminState();
State slaveState = componentHost.getState();
//Delete hostcomponents
- serviceComponent.deleteServiceComponentHosts(componentHost.getHostName());
+ serviceComponent.deleteServiceComponentHosts(componentHost.getHostName(), deleteMetaData);
// If deleted hostcomponents support decomission and were decommited and stopped
if (AmbariCustomCommandExecutionHelper.masterToSlaveMappingForDecom.containsValue(slave_component_name)
&& desiredAdminState.equals(HostComponentAdminState.DECOMMISSIONED)
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java
index 3c0164c..081737a 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java
@@ -32,10 +32,13 @@ import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Request;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.events.listeners.tasks.TaskStatusListener;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.orm.entities.StageEntityPK;
+import org.apache.ambari.server.topology.LogicalRequest;
+import org.apache.ambari.server.topology.TopologyManager;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
@@ -424,6 +427,36 @@ public class CalculatedStatus {
}
/**
+ * Calculates the status for specified request by id.
+ * @param s_hostRoleCommandDAO is used to retrieve the map of stage-to-summary value objects
+ * @param topologyManager topology manager
+ * @param requestId the request id
+ * @return the calculated status
+ */
+ public static CalculatedStatus statusFromRequest(HostRoleCommandDAO s_hostRoleCommandDAO,
+ TopologyManager topologyManager, Long requestId) {
+ Map<Long, HostRoleCommandStatusSummaryDTO> summary = s_hostRoleCommandDAO.findAggregateCounts(requestId);
+
+ // get summaries from TopologyManager for logical requests
+ summary.putAll(topologyManager.getStageSummaries(requestId));
+
+ // summary might be empty due to delete host have cleared all
+ // HostRoleCommands or due to hosts haven't registered yet with the cluster
+ // when the cluster is provisioned with a Blueprint
+ LogicalRequest logicalRequest = topologyManager.getRequest(requestId);
+ if (summary.isEmpty() && null != logicalRequest) {
+ // in this case, it appears that there are no tasks but this is a logical
+ // topology request, so it's a matter of hosts simply not registering yet
+ // for tasks to be created
+ return CalculatedStatus.PENDING;
+ } else {
+ // there are either tasks or this is not a logical request, so do normal
+ // status calculations
+ return CalculatedStatus.statusFromStageSummary(summary, summary.keySet());
+ }
+ }
+
+ /**
* Calculates the overall status of an upgrade. If there are no tasks, then a
* status of {@link HostRoleStatus#COMPLETED} is returned.
*
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
index 65cfcaa..5a56919 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
@@ -61,9 +61,11 @@ import org.apache.ambari.server.state.ServiceComponentFactory;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.State;
+import org.apache.ambari.server.topology.TopologyDeleteFormer;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.Validate;
+import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.AssistedInject;
import com.google.inject.persist.Transactional;
@@ -103,6 +105,9 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
private MaintenanceStateHelper maintenanceStateHelper;
+ @Inject
+ private TopologyDeleteFormer topologyDeleteFormer;
+
// ----- Constructors ----------------------------------------------------
/**
@@ -709,6 +714,7 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
protected RequestStatusResponse deleteComponents(Set<ServiceComponentRequest> requests) throws AmbariException, AuthorizationException {
Clusters clusters = getManagementController().getClusters();
AmbariMetaInfo ambariMetaInfo = getManagementController().getAmbariMetaInfo();
+ DeleteHostComponentStatusMetaData deleteMetaData = new DeleteHostComponentStatusMetaData();
for (ServiceComponentRequest request : requests) {
Validate.notEmpty(request.getComponentName(), "component name should be non-empty");
@@ -721,28 +727,32 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
ServiceComponent sc = s.getServiceComponent(request.getComponentName());
if (sc != null) {
- deleteHostComponentsForServiceComponent(sc, request);
+ deleteHostComponentsForServiceComponent(sc, request, deleteMetaData);
+ topologyDeleteFormer.processDeleteMetaDataException(deleteMetaData);
sc.setDesiredState(State.DISABLED);
- s.deleteServiceComponent(request.getComponentName());
+ s.deleteServiceComponent(request.getComponentName(), deleteMetaData);
+ topologyDeleteFormer.processDeleteMetaDataException(deleteMetaData);
}
}
+ topologyDeleteFormer.processDeleteMetaData(deleteMetaData);
return null;
}
- private void deleteHostComponentsForServiceComponent(ServiceComponent sc, ServiceComponentRequest request) throws AmbariException {
+ private void deleteHostComponentsForServiceComponent(ServiceComponent sc, ServiceComponentRequest request,
+ DeleteHostComponentStatusMetaData deleteMetaData) throws AmbariException {
for (ServiceComponentHost sch : sc.getServiceComponentHosts().values()) {
if (!sch.getDesiredState().isRemovableState()) {
- throw new AmbariException("Found non removable host component when trying to delete service component." +
+ deleteMetaData.setAmbariException(new AmbariException("Found non removable host component when trying to delete service component." +
" To remove host component, it must be in DISABLED/INIT/INSTALLED/INSTALL_FAILED/UNKNOWN" +
"/UNINSTALLED/INSTALLING state."
+ ", request=" + request.toString()
- + ", current state=" + sc.getDesiredState() + ".");
-
+ + ", current state=" + sc.getDesiredState() + "."));
+ return;
}
}
for (ServiceComponentHost sch : sc.getServiceComponentHosts().values()) {
- sch.delete();
+ sch.delete(deleteMetaData);
}
}
private Cluster getClusterForRequest(final ServiceComponentRequest request, final Clusters clusters) throws AmbariException {
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/DeleteHostComponentStatusMetaData.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/DeleteHostComponentStatusMetaData.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/DeleteHostComponentStatusMetaData.java
new file mode 100644
index 0000000..1a4e34a
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/DeleteHostComponentStatusMetaData.java
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.controller.internal;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import org.apache.ambari.server.AmbariException;
+
+@NotThreadSafe
+public class DeleteHostComponentStatusMetaData extends DeleteStatusMetaData {
+ private Set<HostComponent> removedHostComponents;
+ private AmbariException ambariException;
+ public DeleteHostComponentStatusMetaData() {
+ removedHostComponents = new HashSet<>();
+ }
+
+ public void addDeletedHostComponent(String componentName, String hostName, Long hostId,
+ String clusterId, String version) {
+ removedHostComponents.add(new HostComponent(componentName, hostId, clusterId, version));
+ addDeletedKey(componentName + "/" + hostName);
+ }
+
+ public Set<HostComponent> getRemovedHostComponents() {
+ return removedHostComponents;
+ }
+
+ public AmbariException getAmbariException() {
+ return ambariException;
+ }
+
+ public void setAmbariException(AmbariException ambariException) {
+ this.ambariException = ambariException;
+ }
+
+ public class HostComponent {
+ private String componentName;
+ private Long hostId;
+ private String clusterId;
+ private String version;
+
+ public HostComponent(String componentName, Long hostId, String clusterId, String version) {
+ this.componentName = componentName;
+ this.hostId = hostId;
+ this.clusterId = clusterId;
+ this.version = version;
+ }
+
+ public String getComponentName() {
+ return componentName;
+ }
+
+ public void setComponentName(String componentName) {
+ this.componentName = componentName;
+ }
+
+ public Long getHostId() {
+ return hostId;
+ }
+
+ public void setHostName(Long hostId) {
+ this.hostId = hostId;
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(String clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ public void setVersion(String version) {
+ this.version = version;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java
index f604a7f..1b16b5e 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java
@@ -35,6 +35,8 @@ import org.apache.ambari.server.DuplicateResourceException;
import org.apache.ambari.server.HostNotFoundException;
import org.apache.ambari.server.ObjectNotFoundException;
import org.apache.ambari.server.ParentObjectNotFoundException;
+import org.apache.ambari.server.agent.stomp.dto.TopologyCluster;
+import org.apache.ambari.server.agent.stomp.dto.TopologyHost;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.ConfigurationRequest;
import org.apache.ambari.server.controller.HostRequest;
@@ -52,6 +54,8 @@ import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.ambari.server.events.TopologyUpdateEvent;
+import org.apache.ambari.server.events.publishers.StateUpdateEventPublisher;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.security.authorization.AuthorizationHelper;
import org.apache.ambari.server.security.authorization.ResourceType;
@@ -167,6 +171,9 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
@Inject
private static TopologyManager topologyManager;
+ @Inject
+ private StateUpdateEventPublisher stateUpdateEventPublisher;
+
// ----- Constructors ----------------------------------------------------
/**
@@ -508,6 +515,7 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
Map<String, Map<String, String>> hostAttributes = new HashMap<>();
Set<String> allClusterSet = new HashSet<>();
+ Map<String, TopologyCluster> addedTopologies = new HashMap<>();
for (HostRequest hostRequest : hostRequests) {
if (hostRequest.getHostname() != null &&
!hostRequest.getHostname().isEmpty() &&
@@ -521,6 +529,15 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
if (hostRequest.getHostAttributes() != null) {
hostAttributes.put(hostRequest.getHostname(), hostRequest.getHostAttributes());
}
+ String clusterId = Long.toString(clusters.getCluster(hostRequest.getClusterName()).getClusterId());
+ if (!addedTopologies.containsKey(clusterId)) {
+ addedTopologies.put(clusterId, new TopologyCluster());
+ }
+ Host addedHost = clusters.getHost(hostRequest.getHostname());
+ addedTopologies.get(clusterId).addTopologyHost(new TopologyHost(addedHost.getHostId(),
+ addedHost.getHostName(),
+ addedHost.getRackInfo(),
+ addedHost.getIPv4()));
}
}
clusters.updateHostWithClusterAndAttributes(hostClustersMap, hostAttributes);
@@ -528,6 +545,8 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
for (String clusterName : allClusterSet) {
clusters.getCluster(clusterName).recalculateAllClusterVersionStates();
}
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(addedTopologies, TopologyUpdateEvent.EventType.ADD);
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
}
private void createHostResource(Clusters clusters, Set<String> duplicates,
@@ -704,12 +723,14 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
}
}
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
for (HostRequest request : requests) {
if (LOG.isDebugEnabled()) {
LOG.debug("Received an updateHost request"
+ ", hostname=" + request.getHostname()
+ ", request=" + request);
}
+ TopologyHost topologyHost = new TopologyHost();
Host host = clusters.getHost(request.getHostname());
@@ -717,6 +738,7 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
Cluster cluster = clusters.getCluster(clusterName);
Long clusterId = cluster.getClusterId();
Long resourceId = cluster.getResourceId();
+ topologyHost.setHostId(host.getHostId());
try {
// The below method call throws an exception when trying to create a duplicate mapping in the clusterhostmapping
@@ -742,6 +764,7 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
throw new AuthorizationException("The authenticated user is not authorized to update host rack information");
}
host.setRackInfo(requestRackInfo);
+ topologyHost.setRackName(requestRackInfo);
}
if (null != request.getPublicHostName()) {
@@ -749,6 +772,7 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
throw new AuthorizationException("The authenticated user is not authorized to update host attributes");
}
host.setPublicHostName(request.getPublicHostName());
+ topologyHost.setHostName(request.getPublicHostName());
}
if (null != clusterName && null != request.getMaintenanceState()) {
@@ -811,6 +835,13 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
}
}
+ if (!topologyUpdates.containsKey(clusterId.toString())) {
+ topologyUpdates.put(clusterId.toString(), new TopologyCluster());
+ }
+ topologyUpdates.get(clusterId.toString()).addTopologyHost(topologyHost);
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(topologyUpdates,
+ TopologyUpdateEvent.EventType.UPDATE);
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
//todo: if attempt was made to update a property other than those
//todo: that are allowed above, should throw exception
}
@@ -874,9 +905,11 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
Set<String> hostsClusters = new HashSet<>();
Set<String> hostNames = new HashSet<>();
Set<Cluster> allClustersWithHosts = new HashSet<>();
+ Map<String, TopologyCluster> topologyUpdates = new HashMap<>();
for (HostRequest hostRequest : requests) {
// Assume the user also wants to delete it entirely, including all clusters.
String hostname = hostRequest.getHostname();
+ Long hostId = clusters.getHost(hostname).getHostId();
hostNames.add(hostname);
if (hostRequest.getClusterName() != null) {
@@ -911,6 +944,24 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
for (String key : componentDeleteStatus.getDeletedKeys()) {
deleteStatusMetaData.addDeletedKey(key);
}
+ /*for (DeleteHostComponentStatusMetaData.HostComponent hostComponent : componentDeleteStatus.getRemovedHostComponents()) {
+ String clusterId = hostComponent.getClusterId();
+ if (!topologyUpdates.containsKey(clusterId)) {
+ topologyUpdates.put(clusterId, new TopologyCluster());
+ }
+ TopologyComponent deletedComponent = new TopologyComponent(hostComponent.getComponentName(),
+ null,
+ hostComponent.getVersion(),
+ new HashSet<>(Arrays.asList(hostId)),
+ null);
+ if (!topologyUpdates.get(clusterId).getTopologyComponents().contains(deletedComponent)) {
+ topologyUpdates.get(clusterId).addTopologyComponent(deletedComponent);
+ } else {
+ topologyUpdates.get(clusterId).getTopologyComponents()
+ .stream().filter(t -> t.equals(deletedComponent))
+ .forEach(t -> t.addHostId(hostId));
+ }
+ }*/
for (String key : componentDeleteStatus.getExceptionForKeys().keySet()) {
deleteStatusMetaData.addException(key, componentDeleteStatus.getExceptionForKeys().get(key));
}
@@ -920,7 +971,15 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
hostsClusters.add(hostRequest.getClusterName());
}
try {
+ Set<Cluster> hostClusters = new HashSet<>(clusters.getClustersForHost(hostname));
clusters.deleteHost(hostname);
+ for (Cluster cluster : hostClusters) {
+ String clusterId = Long.toString(cluster.getClusterId());
+ if (!topologyUpdates.containsKey(clusterId)) {
+ topologyUpdates.put(clusterId, new TopologyCluster());
+ }
+ topologyUpdates.get(clusterId).getTopologyHosts().add(new TopologyHost(hostId));
+ }
deleteStatusMetaData.addDeletedKey(hostname);
} catch (Exception ex) {
deleteStatusMetaData.addException(hostname, ex);
@@ -934,6 +993,9 @@ public class HostResourceProvider extends AbstractControllerResourceProvider {
for (String clustername : hostsClusters) {
clusters.getCluster(clustername).recalculateAllClusterVersionStates();
}
+ TopologyUpdateEvent topologyUpdateEvent = new TopologyUpdateEvent(topologyUpdates,
+ TopologyUpdateEvent.EventType.DELETE);
+ stateUpdateEventPublisher.publish(topologyUpdateEvent);
}
private void validateHostInDeleteFriendlyState(HostRequest hostRequest, Clusters clusters, boolean forceDelete) throws AmbariException {
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java
index 9704b33..81dc340 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java
@@ -67,7 +67,6 @@ import org.apache.ambari.server.security.authorization.ResourceType;
import org.apache.ambari.server.security.authorization.RoleAuthorization;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
-import org.apache.ambari.server.topology.LogicalRequest;
import org.apache.ambari.server.topology.TopologyManager;
import org.apache.commons.lang.StringUtils;
@@ -738,32 +737,13 @@ public class RequestResourceProvider extends AbstractControllerResourceProvider
setResourceProperty(resource, REQUEST_SOURCE_SCHEDULE, null, requestedPropertyIds);
}
-
- Map<Long, HostRoleCommandStatusSummaryDTO> summary = s_hostRoleCommandDAO.findAggregateCounts(entity.getRequestId());
-
- // get summaries from TopologyManager for logical requests
- summary.putAll(topologyManager.getStageSummaries(entity.getRequestId()));
-
- // summary might be empty due to delete host have cleared all
- // HostRoleCommands or due to hosts haven't registered yet with the cluster
- // when the cluster is provisioned with a Blueprint
- final CalculatedStatus status;
- LogicalRequest logicalRequest = topologyManager.getRequest(entity.getRequestId());
- if (summary.isEmpty() && null != logicalRequest) {
- // in this case, it appears that there are no tasks but this is a logical
- // topology request, so it's a matter of hosts simply not registering yet
- // for tasks to be created
- status = CalculatedStatus.PENDING;
- } else {
- // there are either tasks or this is not a logical request, so do normal
- // status calculations
- status = CalculatedStatus.statusFromStageSummary(summary, summary.keySet());
- }
+ final CalculatedStatus status = CalculatedStatus.statusFromRequest(s_hostRoleCommandDAO, topologyManager, entity.getRequestId());
setResourceProperty(resource, REQUEST_STATUS_PROPERTY_ID, status.getStatus().toString(), requestedPropertyIds);
setResourceProperty(resource, REQUEST_PROGRESS_PERCENT_ID, status.getPercent(), requestedPropertyIds);
int taskCount = 0;
+ Map<Long, HostRoleCommandStatusSummaryDTO> summary = s_hostRoleCommandDAO.findAggregateCounts(entity.getRequestId());
for (HostRoleCommandStatusSummaryDTO dto : summary.values()) {
taskCount += dto.getTaskTotal();
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c2ab4a3c/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
index 6556852..33cab7c 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
@@ -72,6 +72,7 @@ import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.State;
+import org.apache.ambari.server.topology.TopologyDeleteFormer;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.Validate;
@@ -123,6 +124,9 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
@Inject
private KerberosHelper kerberosHelper;
+ @Inject
+ private TopologyDeleteFormer topologyDeleteFormer;
+
// ----- Constructors ----------------------------------------------------
/**
@@ -884,9 +888,12 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
}
}
+ DeleteHostComponentStatusMetaData deleteMetaData = new DeleteHostComponentStatusMetaData();
for (Service service : removable) {
- service.getCluster().deleteService(service.getName());
+ service.getCluster().deleteService(service.getName(), deleteMetaData);
+ topologyDeleteFormer.processDeleteMetaDataException(deleteMetaData);
}
+ topologyDeleteFormer.processDeleteMetaData(deleteMetaData);
return null;
}