HTTP communication channel

cat-core/src/main/java/com/dianping/cat/abtest/ABTestManager.java

@@ -3,6 +3,8 @@ package com.dianping.cat.abtest;
 import javax.servlet.http.HttpServletRequest;
 
 import org.codehaus.plexus.PlexusContainer;
+import org.unidal.helper.Threads;
+import org.unidal.helper.Threads.Task;
 import org.unidal.lookup.ContainerLoader;
 
 import com.dianping.cat.abtest.internal.DefaultABTest;
@@ -25,12 +27,14 @@ public final class ABTestManager {
 					try {
 						// it could be time-consuming due to load entities from the repository, i.e. database.
 						PlexusContainer container = ContainerLoader.getDefaultContainer();
-						
+
 						s_contextManager = container.lookup(ABTestContextManager.class);
-						
+
 						ABTestEntityRepository repository = container.lookup(ABTestEntityRepository.class);
-						
-						repository.start();
+
+						if (repository instanceof Task) {
+							Threads.forGroup("Cat").start((Task) repository);
+						}
 					} catch (Exception e) {
 						throw new RuntimeException("Error when initializing ABTestContextManager!", e);
 					}

cat-core/src/main/java/com/dianping/cat/abtest/repository/ABTestEntityRepository.java

@@ -5,9 +5,5 @@ import java.util.Map;
 import com.dianping.cat.abtest.spi.ABTestEntity;
 
 public interface ABTestEntityRepository {
-
 	public Map<Integer, ABTestEntity> getEntities();
-
-	public void start();
-
 }

cat-core/src/main/java/com/dianping/cat/abtest/repository/DefaultABTestEntityRepository.java

@@ -62,7 +62,6 @@ public class DefaultABTestEntityRepository extends ContainerHolder implements AB
 		m_domain = domain;
 	}
 
-	@Override
 	public void start() {
 		m_socket = new UdpSocket();
 		m_socket.setName("ABTest");

cat-core/src/main/java/com/dianping/cat/abtest/repository/HttpABTestEntityRepository.java

+package com.dianping.cat.abtest.repository;
+
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.LockSupport;
+
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
+import org.unidal.helper.Files;
+import org.unidal.helper.Threads.Task;
+import org.unidal.helper.Urls;
+import org.unidal.lookup.ContainerHolder;
+import org.unidal.lookup.annotation.Inject;
+
+import com.dianping.cat.Cat;
+import com.dianping.cat.abtest.model.entity.AbtestModel;
+import com.dianping.cat.abtest.model.entity.Case;
+import com.dianping.cat.abtest.model.entity.Run;
+import com.dianping.cat.abtest.model.transform.BaseVisitor;
+import com.dianping.cat.abtest.model.transform.DefaultSaxParser;
+import com.dianping.cat.abtest.spi.ABTestEntity;
+import com.dianping.cat.abtest.spi.ABTestGroupStrategy;
+import com.dianping.cat.configuration.ClientConfigManager;
+import com.dianping.cat.configuration.NetworkInterfaceManager;
+import com.dianping.cat.configuration.client.entity.Server;
+import com.dianping.cat.message.Heartbeat;
+import com.dianping.cat.message.Message;
+import com.dianping.cat.message.Transaction;
+
+public class HttpABTestEntityRepository extends ContainerHolder implements ABTestEntityRepository, Initializable, Task {
+
+	@Inject
+	private ClientConfigManager m_configManager;
+
+	private String m_domain;
+
+	private Map<Integer, ABTestEntity> m_entities = new HashMap<Integer, ABTestEntity>();
+
+	private Map<String, ABTestGroupStrategy> m_strategies = new HashMap<String, ABTestGroupStrategy>();
+
+	@Override
+	public Map<Integer, ABTestEntity> getEntities() {
+		return m_entities;
+	}
+
+	@Override
+	public String getName() {
+		return getClass().getSimpleName();
+	}
+
+	@Override
+	public void initialize() throws InitializationException {
+		m_domain = m_configManager.getFirstDomain().getId();
+
+	}
+
+	@Override
+	public void run() {
+		while (true) {
+			long start = System.currentTimeMillis();
+
+			try {
+				refresh();
+			} catch (Throwable e) {
+				Cat.logError(e);
+			}
+
+			LockSupport.parkUntil(start + 6 * 1000L); // every minute
+		}
+	}
+
+	@Override
+	public void shutdown() {
+	}
+
+	private void refresh() {
+		String clientIp = NetworkInterfaceManager.INSTANCE.getLocalHostAddress();
+
+		for (Server server : m_configManager.getServers()) {
+			String ip = server.getIp();
+			int port = server.getHttpPort();
+			String url = String.format("http://%s:%s/cat/s/abtest?op=model", ip, port);
+			Transaction t = Cat.newTransaction("ABTest", url);
+
+			try {
+				InputStream inputStream = Urls.forIO().connectTimeout(100).readTimeout(100).openStream(url);
+				String content = Files.forIO().readFrom(inputStream, "utf-8");
+				AbtestModel abtest = DefaultSaxParser.parse(content);
+				ABTestVisitor visitor = new ABTestVisitor(m_domain);
+
+				abtest.accept(visitor);
+
+				// switch the entities
+				m_entities = visitor.getEntities();
+
+				Heartbeat h = Cat.newHeartbeat("abtest-heartbeat", clientIp);
+
+				h.addData(abtest.toString());
+				h.setStatus(Message.SUCCESS);
+				h.complete();
+
+				t.setStatus(Message.SUCCESS);
+				break;
+			} catch (Throwable e) {
+				t.setStatus(e);
+				Cat.logError(e);
+			} finally {
+				t.complete();
+			}
+		}
+	}
+
+	class ABTestVisitor extends BaseVisitor {
+		private String m_domain;
+
+		private Map<Integer, ABTestEntity> m_entities;
+
+		public ABTestVisitor(String domain) {
+			m_domain = domain;
+			m_entities = new HashMap<Integer, ABTestEntity>();
+		}
+
+		public Map<Integer, ABTestEntity> getEntities() {
+			return m_entities;
+		}
+
+		@Override
+		public void visitCase(Case _case) {
+			for (Run run : _case.getRuns()) {
+				// filter abtest-entities by domain
+				if (run.getDomains() != null && run.getDomains().contains(m_domain)) {
+
+					ABTestEntity entity = new ABTestEntity(_case, run);
+
+					String strategyKey = String.format("%s+%s+%s", _case.getId(), entity.getGroupStrategyName(),
+					      entity.getGroupStrategyConfiguration());
+					ABTestGroupStrategy strategy = m_strategies.get(strategyKey);
+
+					if (strategy != null) {
+						entity.setGroupStrategy(strategy);
+					} else {
+						try {
+							strategy = lookup(ABTestGroupStrategy.class, entity.getGroupStrategyName());
+							entity.setGroupStrategy(strategy);
+
+							m_strategies.put(strategyKey, strategy);
+						} catch (Exception e) {
+							Cat.logError(e);
+
+							ABTestEntity origin = HttpABTestEntityRepository.this.m_entities.get(_case.getId());
+
+							if (origin != null) {
+								entity = origin;
+							} else {
+								entity.setDisabled(true);
+							}
+						}
+					}
+
+					m_entities.put(entity.getId(), entity);
+				}
+			}
+		}
+	}
+
+}

cat-core/src/main/java/com/dianping/cat/build/ABTestComponentConfigurator.java

@@ -7,7 +7,7 @@ import org.unidal.lookup.configuration.AbstractResourceConfigurator;
 import org.unidal.lookup.configuration.Component;
 
 import com.dianping.cat.abtest.repository.ABTestEntityRepository;
-import com.dianping.cat.abtest.repository.DefaultABTestEntityRepository;
+import com.dianping.cat.abtest.repository.HttpABTestEntityRepository;
 import com.dianping.cat.abtest.spi.internal.ABTestContextManager;
 import com.dianping.cat.abtest.spi.internal.ABTestEntityManager;
 import com.dianping.cat.abtest.spi.internal.DefaultABTestContextManager;
@@ -22,9 +22,8 @@ class ABTestComponentConfigurator extends AbstractResourceConfigurator {
 		all.add(C(ABTestContextManager.class, DefaultABTestContextManager.class) //
 		      .req(ABTestEntityManager.class));
 
-		all.add(C(ABTestEntityRepository.class, DefaultABTestEntityRepository.class) //
-		      .req(ClientConfigManager.class) //
-		      .config(E("address").value("228.0.0.3:2283")));
+		all.add(C(ABTestEntityRepository.class, HttpABTestEntityRepository.class) //
+				.req(ClientConfigManager.class));
 
 		all.add(C(ABTestEntityManager.class, DefaultABTestEntityManager.class) //
 				.req(ABTestEntityRepository.class));

cat-core/src/main/resources/META-INF/dal/model/client-codegen.xml

@@ -12,6 +12,7 @@
   <entity name="server">
     <attribute name="ip" value-type="String" />
     <attribute name="port" value-type="int" />
+    <attribute name="http-port" value-type="int" />
     <attribute name="enabled" value-type="boolean" />
   </entity>
   <entity name="domain">

cat-core/src/main/resources/META-INF/dal/model/client-model.xml

@@ -11,6 +11,8 @@
 	</entity>
 	<entity name="server">
 		<attribute name="ip" value-type="String" key="true" />
+	    <attribute name="port" value-type="int" default-value="2280" />
+	    <attribute name="http-port" value-type="int" default-value="2281" />
     	<attribute name="enabled" value-type="boolean" default-value="true"/>
 	</entity>
 	<entity name="domain">

cat-core/src/main/resources/META-INF/plexus/components.xml

@@ -313,10 +313,7 @@
 		</component>
 		<component>
 			<role>com.dianping.cat.abtest.repository.ABTestEntityRepository</role>
-			<implementation>com.dianping.cat.abtest.repository.DefaultABTestEntityRepository</implementation>
-			<configuration>
-				<address>228.0.0.3:2283</address>
-			</configuration>
+			<implementation>com.dianping.cat.abtest.repository.HttpABTestEntityRepository</implementation>
 			<requirements>
 				<requirement>
 					<role>com.dianping.cat.configuration.ClientConfigManager</role>

cat-core/src/main/resources/com/dianping/cat/configuration/client/config.xsd

@@ -30,7 +30,8 @@
       <xs:simpleContent>
          <xs:extension base="xs:string">
             <xs:attribute name="ip" type="xs:string" use="required"/>
-            <xs:attribute name="port" type="xs:int"/>
+            <xs:attribute name="port" type="xs:int" default="2280"/>
+            <xs:attribute name="http-port" type="xs:int" default="2281"/>
             <xs:attribute name="enabled" type="xs:boolean" default="true"/>
          </xs:extension>
       </xs:simpleContent>

cat-home/src/main/java/com/dianping/cat/system/page/abtest/Action.java

@@ -7,6 +7,8 @@ public enum Action implements org.unidal.web.mvc.Action {
 	
 	DETAIL("detail"),
 	
+	MODEL("model"),
+	
 	REPORT("report");
 
 	private String m_name;

cat-home/src/main/java/com/dianping/cat/system/page/abtest/Handler.java

@@ -24,6 +24,9 @@ import com.dainping.cat.consumer.dal.report.Project;
 import com.dainping.cat.consumer.dal.report.ProjectDao;
 import com.dainping.cat.consumer.dal.report.ProjectEntity;
 import com.dianping.cat.Cat;
+import com.dianping.cat.abtest.model.entity.AbtestModel;
+import com.dianping.cat.abtest.model.entity.Case;
+import com.dianping.cat.abtest.model.entity.Run;
 import com.dianping.cat.home.dal.abtest.Abtest;
 import com.dianping.cat.home.dal.abtest.AbtestDao;
 import com.dianping.cat.home.dal.abtest.AbtestEntity;
@@ -212,6 +215,9 @@ public class Handler implements PageHandler<Context>, LogEnabled {
 		case REPORT:
 			renderReportModel(ctx, model, payload);
 			break;
+		case MODEL:
+			renderModel(model);
+			break;
 		}
 
 		model.setAction(action);
@@ -219,6 +225,69 @@ public class Handler implements PageHandler<Context>, LogEnabled {
 		m_jspViewer.view(ctx, model);
 	}
 
+	private void renderModel(Model model) {
+		model.setAbtestModel(fetchAbtestModel().toString());
+   }
+	
+	private AbtestModel fetchAbtestModel() {
+		try {
+			AbtestModel abtestModel = new AbtestModel();
+
+			List<AbtestRun> abtestRuns = m_abtestRunDao.findAll(AbtestRunEntity.READSET_FULL);
+
+			if (abtestRuns != null) {
+				Date now = new Date();
+				for (AbtestRun abtestRun : abtestRuns) {
+					AbtestStatus status = AbtestStatus.calculateStatus(abtestRun, now);
+					if (status == AbtestStatus.READY || status == AbtestStatus.RUNNING) {
+						// fetch Case and GroupStrategy
+						int caseId = abtestRun.getCaseId();
+						Abtest entity = m_abtestDao.findByPK(caseId, AbtestEntity.READSET_FULL);
+						int gid = entity.getGroupStrategy();
+						GroupStrategy groupStrategy = m_groupStrategyDao.findByPK(gid, GroupStrategyEntity.READSET_FULL);
+
+						Case _case = transform(abtestRun, entity, groupStrategy);
+						abtestModel.addCase(_case);
+
+					}
+				}
+			}
+
+			return abtestModel;
+		} catch (DalException e) {
+			m_logger.error("Error when find all AbtestRun", e);
+			Cat.logError(e);
+		}
+		return null;
+	}
+
+	private Case transform(AbtestRun abtestRun, Abtest entity, GroupStrategy groupStrategy) throws DalException {
+		Case _case = new Case(entity.getId());
+		_case.setCreatedDate(entity.getCreationDate());
+		_case.setDescription(entity.getDescription());
+		_case.setGroupStrategy(groupStrategy.getName());
+		_case.setName(entity.getName());
+		_case.setOwner(entity.getOwner());
+		_case.setLastModifiedDate(entity.getModifiedDate());
+		for (String domain : StringUtils.split(entity.getDomains(), ',')) {
+			_case.addDomain(domain);
+		}
+
+		Run run = new Run(abtestRun.getId());
+		for (String domain : StringUtils.split(abtestRun.getDomains(), ',')) {
+			run.addDomain(domain);
+		}
+		run.setCreator(abtestRun.getCreator());
+		run.setDisabled(false);
+		run.setEndDate(abtestRun.getEndDate());
+		run.setGroupStrategyConfiguration(abtestRun.getStrategyConfiguration());
+		run.setStartDate(abtestRun.getStartDate());
+
+		_case.addRun(run);
+
+		return _case;
+	}
+
 	private void renderCreateModel(Model model) {
 		Map<String, List<Project>> projectMap = getAllProjects();
 		List<GroupStrategy> groupStrategyList = getAllGroupStrategys();

cat-home/src/main/java/com/dianping/cat/system/page/abtest/JspFile.java

@@ -8,7 +8,9 @@ public enum JspFile {
 	
 	VIEW("/jsp/system/abtestAllTest.jsp"),
 	
-	REPORT("/jsp/system/abtestReport.jsp")
+	REPORT("/jsp/system/abtestReport.jsp"),
+	
+	MODEL("/jsp/system/abtestModel.jsp")
 	;
 
 	private String m_path;

cat-home/src/main/java/com/dianping/cat/system/page/abtest/JspViewer.java

@@ -17,6 +17,9 @@ public class JspViewer extends BaseJspViewer<SystemPage, Action, Context, Model>
 			return JspFile.DETAIL.getPath();
 		case REPORT:
 			return JspFile.REPORT.getPath();
+		case MODEL:
+			return JspFile.MODEL.getPath();
+			
 		}
 
 		throw new RuntimeException("Unknown action: " + action);

cat-home/src/main/java/com/dianping/cat/system/page/abtest/Model.java

@@ -40,6 +40,8 @@ public class Model extends ViewModel<SystemPage, Action, Context> {
 
 	private AbtestDaoModel m_abtest;
 
+	private String m_abtestModel;
+	
 	public Model(Context ctx) {
 		super(ctx);
 	}
@@ -153,6 +155,14 @@ public class Model extends ViewModel<SystemPage, Action, Context> {
 		m_abtest = abtest;
 	}
 
+	public String getAbtestModel() {
+   	return m_abtestModel;
+   }
+
+	public void setAbtestModel(String abtestModel) {
+   	m_abtestModel = abtestModel;
+   }
+
 	public static class AbtestDaoModel {
 
 		private Abtest m_abtest;

cat-home/src/main/webapp/jsp/system/abtestModel.jsp

+<jsp:useBean id="model" type="com.dianping.cat.system.page.abtest.Model" scope="request" />${model.abtestModel}