001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.camel.management.mbean; 018 019import java.io.InputStream; 020import java.io.Serializable; 021import java.util.ArrayList; 022import java.util.Collection; 023import java.util.Collections; 024import java.util.Comparator; 025import java.util.Date; 026import java.util.HashMap; 027import java.util.List; 028import java.util.Map; 029import java.util.Set; 030import java.util.concurrent.RejectedExecutionException; 031import java.util.concurrent.TimeUnit; 032 033import javax.management.AttributeValueExp; 034import javax.management.MBeanServer; 035import javax.management.ObjectName; 036import javax.management.Query; 037import javax.management.QueryExp; 038import javax.management.StringValueExp; 039import javax.management.openmbean.CompositeData; 040import javax.management.openmbean.CompositeDataSupport; 041import javax.management.openmbean.CompositeType; 042import javax.management.openmbean.TabularData; 043import javax.management.openmbean.TabularDataSupport; 044 045import org.apache.camel.CamelContext; 046import org.apache.camel.ExtendedCamelContext; 047import org.apache.camel.ManagementStatisticsLevel; 048import org.apache.camel.Route; 049import org.apache.camel.RuntimeCamelException; 050import org.apache.camel.ServiceStatus; 051import org.apache.camel.TimerListener; 052import org.apache.camel.api.management.ManagedResource; 053import org.apache.camel.api.management.mbean.CamelOpenMBeanTypes; 054import org.apache.camel.api.management.mbean.ManagedProcessorMBean; 055import org.apache.camel.api.management.mbean.ManagedRouteMBean; 056import org.apache.camel.api.management.mbean.ManagedStepMBean; 057import org.apache.camel.api.management.mbean.RouteError; 058import org.apache.camel.model.Model; 059import org.apache.camel.model.ModelCamelContext; 060import org.apache.camel.model.RouteDefinition; 061import org.apache.camel.model.RoutesDefinition; 062import org.apache.camel.spi.InflightRepository; 063import org.apache.camel.spi.ManagementStrategy; 064import org.apache.camel.spi.RoutePolicy; 065import org.apache.camel.support.PluginHelper; 066import org.apache.camel.util.ObjectHelper; 067import org.apache.camel.util.json.JsonArray; 068import org.apache.camel.util.json.JsonObject; 069import org.apache.camel.xml.LwModelHelper; 070import org.slf4j.Logger; 071import org.slf4j.LoggerFactory; 072 073@ManagedResource(description = "Managed Route") 074public class ManagedRoute extends ManagedPerformanceCounter implements TimerListener, ManagedRouteMBean { 075 076 public static final String VALUE_UNKNOWN = "Unknown"; 077 078 private static final Logger LOG = LoggerFactory.getLogger(ManagedRoute.class); 079 080 protected final Route route; 081 protected final String description; 082 protected final String note; 083 protected final String configurationId; 084 protected final String sourceLocation; 085 protected final String sourceLocationShort; 086 protected final CamelContext context; 087 private final LoadTriplet load = new LoadTriplet(); 088 private final LoadThroughput thp = new LoadThroughput(); 089 private final String jmxDomain; 090 091 public ManagedRoute(CamelContext context, Route route) { 092 this.route = route; 093 this.context = context; 094 this.description = route.getDescription(); 095 this.note = route.getNote(); 096 this.configurationId = route.getConfigurationId(); 097 this.sourceLocation = route.getSourceLocation(); 098 this.sourceLocationShort = route.getSourceLocationShort(); 099 this.jmxDomain = context.getManagementStrategy().getManagementAgent().getMBeanObjectDomainName(); 100 } 101 102 @Override 103 public void init(ManagementStrategy strategy) { 104 super.init(strategy); 105 boolean enabled 106 = context.getManagementStrategy().getManagementAgent().getStatisticsLevel() != ManagementStatisticsLevel.Off; 107 setStatisticsEnabled(enabled); 108 } 109 110 public Route getRoute() { 111 return route; 112 } 113 114 public CamelContext getContext() { 115 return context; 116 } 117 118 @Override 119 public String getRouteId() { 120 String id = route.getId(); 121 if (id == null) { 122 id = VALUE_UNKNOWN; 123 } 124 return id; 125 } 126 127 @Override 128 public String getNodePrefixId() { 129 return route.getNodePrefixId(); 130 } 131 132 @Override 133 public String getRouteGroup() { 134 return route.getGroup(); 135 } 136 137 @Override 138 public boolean isCreatedByRouteTemplate() { 139 return "true".equals(route.getProperties().getOrDefault(Route.TEMPLATE_PROPERTY, "false")); 140 } 141 142 @Override 143 public boolean isCreatedByKamelet() { 144 return "true".equals(route.getProperties().getOrDefault(Route.KAMELET_PROPERTY, "false")); 145 } 146 147 @Override 148 public TabularData getRouteProperties() { 149 try { 150 final Map<String, Object> properties = route.getProperties(); 151 final TabularData answer = new TabularDataSupport(CamelOpenMBeanTypes.camelRoutePropertiesTabularType()); 152 final CompositeType ct = CamelOpenMBeanTypes.camelRoutePropertiesCompositeType(); 153 154 // gather route properties 155 for (Map.Entry<String, Object> entry : properties.entrySet()) { 156 final String key = entry.getKey(); 157 final String val = context.getTypeConverter().convertTo(String.class, entry.getValue()); 158 159 CompositeData data = new CompositeDataSupport( 160 ct, 161 new String[] { "key", "value" }, 162 new Object[] { key, val }); 163 164 answer.put(data); 165 } 166 return answer; 167 } catch (Exception e) { 168 throw RuntimeCamelException.wrapRuntimeCamelException(e); 169 } 170 } 171 172 @Override 173 public String getDescription() { 174 return description; 175 } 176 177 @Override 178 public String getNote() { 179 return note; 180 } 181 182 @Override 183 public Boolean getAutoStartup() { 184 return route.isAutoStartup(); 185 } 186 187 @Override 188 public String getSourceLocation() { 189 return sourceLocation; 190 } 191 192 @Override 193 public String getSourceLocationShort() { 194 return sourceLocationShort; 195 } 196 197 @Override 198 public String getRouteConfigurationId() { 199 return configurationId; 200 } 201 202 @Override 203 public String getEndpointUri() { 204 if (route.getEndpoint() != null) { 205 return route.getEndpoint().getEndpointUri(); 206 } 207 return VALUE_UNKNOWN; 208 } 209 210 @Override 211 public String getState() { 212 // must use String type to be sure remote JMX can read the attribute without requiring Camel classes. 213 ServiceStatus status = context.getRouteController().getRouteStatus(route.getId()); 214 // if no status exists then its stopped 215 if (status == null) { 216 status = ServiceStatus.Stopped; 217 } 218 return status.name(); 219 } 220 221 @Override 222 public String getUptime() { 223 return route.getUptime(); 224 } 225 226 @Override 227 public long getUptimeMillis() { 228 return route.getUptimeMillis(); 229 } 230 231 @Override 232 public String getCamelId() { 233 return context.getName(); 234 } 235 236 @Override 237 public String getCamelManagementName() { 238 return context.getManagementName(); 239 } 240 241 @Override 242 public Boolean getTracing() { 243 return route.isTracing(); 244 } 245 246 @Override 247 public void setTracing(Boolean tracing) { 248 route.setTracing(tracing); 249 } 250 251 @Override 252 public Boolean getMessageHistory() { 253 return route.isMessageHistory(); 254 } 255 256 @Override 257 public Boolean getLogMask() { 258 return route.isLogMask(); 259 } 260 261 @Override 262 public String getRoutePolicyList() { 263 List<RoutePolicy> policyList = route.getRoutePolicyList(); 264 265 if (policyList == null || policyList.isEmpty()) { 266 // return an empty string to have it displayed nicely in JMX consoles 267 return ""; 268 } 269 270 StringBuilder sb = new StringBuilder(); 271 for (int i = 0; i < policyList.size(); i++) { 272 RoutePolicy policy = policyList.get(i); 273 sb.append(policy.getClass().getSimpleName()); 274 sb.append("(").append(ObjectHelper.getIdentityHashCode(policy)).append(")"); 275 if (i < policyList.size() - 1) { 276 sb.append(", "); 277 } 278 } 279 return sb.toString(); 280 } 281 282 @Override 283 public String getLoad01() { 284 double load1 = load.getLoad1(); 285 if (Double.isNaN(load1)) { 286 // empty string if load statistics is disabled 287 return ""; 288 } else { 289 return String.format("%.2f", load1); 290 } 291 } 292 293 @Override 294 public String getLoad05() { 295 double load5 = load.getLoad5(); 296 if (Double.isNaN(load5)) { 297 // empty string if load statistics is disabled 298 return ""; 299 } else { 300 return String.format("%.2f", load5); 301 } 302 } 303 304 @Override 305 public String getLoad15() { 306 double load15 = load.getLoad15(); 307 if (Double.isNaN(load15)) { 308 // empty string if load statistics is disabled 309 return ""; 310 } else { 311 return String.format("%.2f", load15); 312 } 313 } 314 315 @Override 316 public String getThroughput() { 317 double d = thp.getThroughput(); 318 if (Double.isNaN(d)) { 319 // empty string if load statistics is disabled 320 return ""; 321 } else { 322 return String.format("%.2f", d); 323 } 324 } 325 326 @Override 327 public void onTimer() { 328 load.update(getInflightExchanges()); 329 thp.update(getExchangesTotal()); 330 } 331 332 @Override 333 public void start() throws Exception { 334 if (!context.getStatus().isStarted()) { 335 throw new IllegalArgumentException("CamelContext is not started"); 336 } 337 try { 338 context.getRouteController().startRoute(getRouteId()); 339 } catch (Exception e) { 340 LOG.warn("Error starting route: {} due to: {}. This exception is ignored.", getRouteId(), e.getMessage(), e); 341 throw e; 342 } 343 } 344 345 @Override 346 public void stop() throws Exception { 347 if (!context.getStatus().isStarted()) { 348 throw new IllegalArgumentException("CamelContext is not started"); 349 } 350 try { 351 context.getRouteController().stopRoute(getRouteId()); 352 } catch (Exception e) { 353 LOG.warn("Error stopping route: {} due to: {}. This exception is ignored.", getRouteId(), e.getMessage(), e); 354 throw e; 355 } 356 } 357 358 @Override 359 public void stopAndFail() throws Exception { 360 if (!context.getStatus().isStarted()) { 361 throw new IllegalArgumentException("CamelContext is not started"); 362 } 363 Throwable cause = new RejectedExecutionException("Route " + getRouteId() + " is forced stopped and marked as failed"); 364 context.getRouteController().stopRoute(getRouteId(), cause); 365 } 366 367 @Override 368 public void stop(long timeout) throws Exception { 369 if (!context.getStatus().isStarted()) { 370 throw new IllegalArgumentException("CamelContext is not started"); 371 } 372 context.getRouteController().stopRoute(getRouteId(), timeout, TimeUnit.SECONDS); 373 } 374 375 @Override 376 public boolean stop(Long timeout, Boolean abortAfterTimeout) throws Exception { 377 if (!context.getStatus().isStarted()) { 378 throw new IllegalArgumentException("CamelContext is not started"); 379 } 380 return context.getRouteController().stopRoute(getRouteId(), timeout, TimeUnit.SECONDS, abortAfterTimeout); 381 } 382 383 /** 384 * @deprecated not in use 385 */ 386 @Deprecated(since = "4.8.0") 387 public void shutdown() throws Exception { 388 if (!context.getStatus().isStarted()) { 389 throw new IllegalArgumentException("CamelContext is not started"); 390 } 391 String routeId = getRouteId(); 392 context.getRouteController().stopRoute(routeId); 393 context.removeRoute(routeId); 394 } 395 396 /** 397 * @deprecated not in use 398 */ 399 @Deprecated(since = "4.8.0") 400 public void shutdown(long timeout) throws Exception { 401 if (!context.getStatus().isStarted()) { 402 throw new IllegalArgumentException("CamelContext is not started"); 403 } 404 String routeId = getRouteId(); 405 context.getRouteController().stopRoute(routeId, timeout, TimeUnit.SECONDS); 406 context.removeRoute(routeId); 407 } 408 409 @Override 410 public boolean remove() throws Exception { 411 if (!context.getStatus().isStarted()) { 412 throw new IllegalArgumentException("CamelContext is not started"); 413 } 414 return context.removeRoute(getRouteId()); 415 } 416 417 @Override 418 public void restart() throws Exception { 419 restart(1); 420 } 421 422 @Override 423 public void restart(long delay) throws Exception { 424 stop(); 425 if (delay > 0) { 426 try { 427 LOG.debug("Sleeping {} seconds before starting route: {}", delay, getRouteId()); 428 Thread.sleep(delay * 1000); 429 } catch (InterruptedException e) { 430 LOG.info("Interrupted while waiting before starting the route"); 431 Thread.currentThread().interrupt(); 432 } 433 } 434 start(); 435 } 436 437 @Override 438 public String dumpRouteAsXml() throws Exception { 439 return dumpRouteAsXml(false); 440 } 441 442 @Override 443 public String dumpRouteAsXml(boolean resolvePlaceholders) throws Exception { 444 return dumpRouteAsXml(resolvePlaceholders, true); 445 } 446 447 @Override 448 public String dumpRouteAsXml(boolean resolvePlaceholders, boolean generatedIds) throws Exception { 449 return dumpRouteAsXml(resolvePlaceholders, true, false); 450 } 451 452 @Override 453 public String dumpRouteAsXml(boolean resolvePlaceholders, boolean generatedIds, boolean sourceLocation) throws Exception { 454 String id = route.getId(); 455 RouteDefinition def = context.getCamelContextExtension().getContextPlugin(Model.class).getRouteDefinition(id); 456 if (def != null) { 457 // if we are debugging then ids is needed for the debugger 458 if (context.isDebugging()) { 459 generatedIds = true; 460 } 461 return PluginHelper.getModelToXMLDumper(context).dumpModelAsXml(context, def, resolvePlaceholders, generatedIds, 462 sourceLocation); 463 } 464 465 return null; 466 } 467 468 @Override 469 public String dumpRouteAsYaml() throws Exception { 470 return dumpRouteAsYaml(false, false); 471 } 472 473 @Override 474 public String dumpRouteAsYaml(boolean resolvePlaceholders) throws Exception { 475 return dumpRouteAsYaml(resolvePlaceholders, false); 476 } 477 478 @Override 479 public String dumpRouteAsYaml(boolean resolvePlaceholders, boolean uriAsParameters) throws Exception { 480 return dumpRouteAsYaml(resolvePlaceholders, uriAsParameters, true, false); 481 } 482 483 @Override 484 public String dumpRouteAsYaml( 485 boolean resolvePlaceholders, boolean uriAsParameters, boolean generatedIds, boolean sourceLocation) 486 throws Exception { 487 String id = route.getId(); 488 RouteDefinition def = context.getCamelContextExtension().getContextPlugin(Model.class).getRouteDefinition(id); 489 if (def != null) { 490 return PluginHelper.getModelToYAMLDumper(context).dumpModelAsYaml(context, def, resolvePlaceholders, 491 uriAsParameters, generatedIds, sourceLocation); 492 } 493 494 return null; 495 } 496 497 @Override 498 public String dumpRouteStatsAsXml(boolean fullStats, boolean includeProcessors) throws Exception { 499 // in this logic we need to calculate the accumulated processing time for the processor in the route 500 // and hence why the logic is a bit more complicated to do this, as we need to calculate that from 501 // the bottom -> top of the route but this information is valuable for profiling routes 502 StringBuilder sb = new StringBuilder(); 503 504 // need to calculate this value first, as we need that value for the route stat 505 long processorAccumulatedTime = 0L; 506 507 // gather all the processors for this route, which requires JMX 508 if (includeProcessors) { 509 sb.append(" <processorStats>\n"); 510 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 511 if (server != null) { 512 // get all the processor mbeans and sort them accordingly to their index 513 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 514 ObjectName query = ObjectName.getInstance( 515 jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*"); 516 Set<ObjectName> names = server.queryNames(query, null); 517 List<ManagedProcessorMBean> mps = new ArrayList<>(); 518 for (ObjectName on : names) { 519 ManagedProcessorMBean processor = context.getManagementStrategy().getManagementAgent().newProxyClient(on, 520 ManagedProcessorMBean.class); 521 522 // the processor must belong to this route 523 if (getRouteId().equals(processor.getRouteId())) { 524 mps.add(processor); 525 } 526 } 527 mps.sort(new OrderProcessorMBeans()); 528 529 // walk the processors in reverse order, and calculate the accumulated total time 530 Map<String, Long> accumulatedTimes = new HashMap<>(); 531 Collections.reverse(mps); 532 for (ManagedProcessorMBean processor : mps) { 533 processorAccumulatedTime += processor.getTotalProcessingTime(); 534 accumulatedTimes.put(processor.getProcessorId(), processorAccumulatedTime); 535 } 536 // and reverse back again 537 Collections.reverse(mps); 538 539 // and now add the sorted list of processors to the xml output 540 for (ManagedProcessorMBean processor : mps) { 541 int line = processor.getSourceLineNumber() != null ? processor.getSourceLineNumber() : -1; 542 sb.append(" <processorStat") 543 .append(String.format( 544 " id=\"%s\" index=\"%s\" state=\"%s\" disabled=\"%s\" sourceLineNumber=\"%s\"", 545 processor.getProcessorId(), processor.getIndex(), processor.getState(), 546 processor.getDisabled(), line)); 547 // do we have an accumulated time then append that 548 Long accTime = accumulatedTimes.get(processor.getProcessorId()); 549 if (accTime != null) { 550 sb.append(" accumulatedProcessingTime=\"").append(accTime).append("\""); 551 } 552 // use substring as we only want the attributes 553 sb.append(" ").append(processor.dumpStatsAsXml(fullStats).substring(7)).append("\n"); 554 } 555 } 556 sb.append(" </processorStats>\n"); 557 } 558 559 // route self time is route total - processor accumulated total) 560 long routeSelfTime = getTotalProcessingTime() - processorAccumulatedTime; 561 if (routeSelfTime < 0) { 562 // ensure we don't calculate that as negative 563 routeSelfTime = 0; 564 } 565 566 StringBuilder answer = new StringBuilder(); 567 answer.append("<routeStat").append(String.format(" id=\"%s\"", route.getId())) 568 .append(String.format(" state=\"%s\"", getState())) 569 .append(String.format(" uptime=\"%s\"", getUptimeMillis())); 570 if (getRouteGroup() != null) { 571 answer.append(String.format(" group=\"%s\"", getRouteGroup())); 572 } 573 if (sourceLocation != null) { 574 answer.append(String.format(" sourceLocation=\"%s\"", getSourceLocation())); 575 } 576 // use substring as we only want the attributes 577 String stat = dumpStatsAsXml(fullStats); 578 answer.append(" exchangesInflight=\"").append(getInflightExchanges()).append("\""); 579 answer.append(" selfProcessingTime=\"").append(routeSelfTime).append("\""); 580 InflightRepository.InflightExchange oldest = getOldestInflightEntry(); 581 if (oldest == null) { 582 answer.append(" oldestInflightExchangeId=\"\""); 583 answer.append(" oldestInflightDuration=\"\""); 584 } else { 585 answer.append(" oldestInflightExchangeId=\"").append(oldest.getExchange().getExchangeId()).append("\""); 586 answer.append(" oldestInflightDuration=\"").append(oldest.getDuration()).append("\""); 587 } 588 answer.append(" ").append(stat, 7, stat.length() - 2).append(">\n"); 589 590 if (includeProcessors) { 591 answer.append(sb); 592 } 593 594 answer.append("</routeStat>"); 595 return answer.toString(); 596 } 597 598 @Override 599 public String dumpRouteStatsAsJSon(boolean fullStats, boolean includeProcessors) throws Exception { 600 // in this logic we need to calculate the accumulated processing time for the processor in the route 601 // and hence why the logic is a bit more complicated to do this, as we need to calculate that from 602 // the bottom -> top of the route but this information is valuable for profiling routes 603 JsonObject root = new JsonObject(); 604 605 root.put("id", getRouteId()); 606 root.put("state", getState()); 607 root.put("uptime", getUptimeMillis()); 608 if (getRouteGroup() != null) { 609 root.put("group", getRouteGroup()); 610 } 611 if (sourceLocation != null) { 612 root.put("sourceLocation", sourceLocation); 613 } 614 statsAsJSon(root, fullStats); 615 root.put("exchangesInflight", getInflightExchanges()); 616 InflightRepository.InflightExchange oldest = getOldestInflightEntry(); 617 if (oldest != null) { 618 root.put("oldestInflightExchangeId", oldest.getExchange().getExchangeId()); 619 root.put("oldestInflightDuration", oldest.getDuration()); 620 } 621 622 // need to calculate this value first, as we need that value for the route stat 623 long processorAccumulatedTime = 0L; 624 625 // gather all the processors for this route, which requires JMX 626 JsonArray arr = null; 627 if (includeProcessors) { 628 arr = new JsonArray(); 629 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 630 if (server != null) { 631 // get all the processor mbeans and sort them accordingly to their index 632 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 633 ObjectName query = ObjectName.getInstance( 634 jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*"); 635 Set<ObjectName> names = server.queryNames(query, null); 636 List<ManagedProcessorMBean> mps = new ArrayList<>(); 637 for (ObjectName on : names) { 638 ManagedProcessorMBean processor = context.getManagementStrategy().getManagementAgent().newProxyClient(on, 639 ManagedProcessorMBean.class); 640 641 // the processor must belong to this route 642 if (getRouteId().equals(processor.getRouteId())) { 643 mps.add(processor); 644 } 645 } 646 mps.sort(new OrderProcessorMBeans()); 647 648 // walk the processors in reverse order, and calculate the accumulated total time 649 Map<String, Long> accumulatedTimes = new HashMap<>(); 650 Collections.reverse(mps); 651 for (ManagedProcessorMBean processor : mps) { 652 processorAccumulatedTime += processor.getTotalProcessingTime(); 653 accumulatedTimes.put(processor.getProcessorId(), processorAccumulatedTime); 654 } 655 // and reverse back again 656 Collections.reverse(mps); 657 658 // and now add the sorted list of processors to the xml output 659 for (ManagedProcessorMBean processor : mps) { 660 JsonObject jo = new JsonObject(); 661 arr.add(jo); 662 processor.statsAsJSon(jo, fullStats); 663 664 int line = processor.getSourceLineNumber() != null ? processor.getSourceLineNumber() : -1; 665 jo.put("id", processor.getProcessorId()); 666 jo.put("index", processor.getIndex()); 667 jo.put("state", processor.getState()); 668 jo.put("disabled", processor.getDisabled()); 669 jo.put("sourceLineNumber", line); 670 671 // do we have an accumulated time then append that 672 Long accTime = accumulatedTimes.get(processor.getProcessorId()); 673 if (accTime != null) { 674 jo.put("accumulatedProcessingTime", accTime); 675 } 676 } 677 } 678 } 679 680 // route self time is route total - processor accumulated total 681 long routeSelfTime = getTotalProcessingTime() - processorAccumulatedTime; 682 if (routeSelfTime < 0) { 683 // ensure we don't calculate that as negative 684 routeSelfTime = 0; 685 } 686 root.put("selfProcessingTime", routeSelfTime); 687 if (arr != null) { 688 // processors should be last 689 root.put("processors", arr); 690 } 691 692 return root.toJson(); 693 } 694 695 @Override 696 public String dumpStepStatsAsXml(boolean fullStats) throws Exception { 697 // in this logic we need to calculate the accumulated processing time for the processor in the route 698 // and hence why the logic is a bit more complicated to do this, as we need to calculate that from 699 // the bottom -> top of the route but this information is valuable for profiling routes 700 StringBuilder sb = new StringBuilder(); 701 702 // gather all the steps for this route, which requires JMX 703 sb.append(" <stepStats>\n"); 704 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 705 if (server != null) { 706 // get all the processor mbeans and sort them accordingly to their index 707 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 708 ObjectName query = ObjectName 709 .getInstance(jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=steps,*"); 710 Set<ObjectName> names = server.queryNames(query, null); 711 List<ManagedStepMBean> mps = new ArrayList<>(); 712 for (ObjectName on : names) { 713 ManagedStepMBean step 714 = context.getManagementStrategy().getManagementAgent().newProxyClient(on, ManagedStepMBean.class); 715 716 // the step must belong to this route 717 if (getRouteId().equals(step.getRouteId())) { 718 mps.add(step); 719 } 720 } 721 mps.sort(new OrderProcessorMBeans()); 722 723 // and now add the sorted list of steps to the xml output 724 for (ManagedStepMBean step : mps) { 725 int line = step.getSourceLineNumber() != null ? step.getSourceLineNumber() : -1; 726 sb.append(" <stepStat") 727 .append(String.format(" id=\"%s\" index=\"%s\" state=\"%s\" sourceLineNumber=\"%s\"", 728 step.getProcessorId(), 729 step.getIndex(), step.getState(), line)); 730 // use substring as we only want the attributes 731 sb.append(" ").append(step.dumpStatsAsXml(fullStats).substring(7)).append("\n"); 732 } 733 } 734 sb.append(" </stepStats>\n"); 735 736 StringBuilder answer = new StringBuilder(); 737 answer.append("<routeStat").append(String.format(" id=\"%s\"", route.getId())) 738 .append(String.format(" state=\"%s\"", getState())) 739 .append(String.format(" uptime=\"%s\"", getUptimeMillis())); 740 if (getRouteGroup() != null) { 741 answer.append(String.format(" group=\"%s\"", getRouteGroup())); 742 } 743 if (sourceLocation != null) { 744 answer.append(String.format(" sourceLocation=\"%s\"", getSourceLocation())); 745 } 746 // use substring as we only want the attributes 747 String stat = dumpStatsAsXml(fullStats); 748 answer.append(" exchangesInflight=\"").append(getInflightExchanges()).append("\""); 749 InflightRepository.InflightExchange oldest = getOldestInflightEntry(); 750 if (oldest == null) { 751 answer.append(" oldestInflightExchangeId=\"\""); 752 answer.append(" oldestInflightDuration=\"\""); 753 } else { 754 answer.append(" oldestInflightExchangeId=\"").append(oldest.getExchange().getExchangeId()).append("\""); 755 answer.append(" oldestInflightDuration=\"").append(oldest.getDuration()).append("\""); 756 } 757 answer.append(" ").append(stat, 7, stat.length() - 2).append(">\n"); 758 759 answer.append(sb); 760 761 answer.append("</routeStat>"); 762 return answer.toString(); 763 } 764 765 @Override 766 public String dumpRouteSourceLocationsAsXml() throws Exception { 767 StringBuilder sb = new StringBuilder(); 768 sb.append("<routeLocations>"); 769 770 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 771 if (server != null) { 772 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 773 List<ManagedProcessorMBean> processors = new ArrayList<>(); 774 // gather all the processors for this CamelContext, which requires JMX 775 ObjectName query = ObjectName 776 .getInstance(jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*"); 777 Set<ObjectName> names = server.queryNames(query, null); 778 for (ObjectName on : names) { 779 ManagedProcessorMBean processor 780 = context.getManagementStrategy().getManagementAgent().newProxyClient(on, ManagedProcessorMBean.class); 781 // the processor must belong to this route 782 if (getRouteId().equals(processor.getRouteId())) { 783 processors.add(processor); 784 } 785 } 786 processors.sort(new OrderProcessorMBeans()); 787 788 // grab route consumer 789 RouteDefinition rd = ((ModelCamelContext) context).getRouteDefinition(route.getRouteId()); 790 if (rd != null) { 791 String id = rd.getRouteId(); 792 int line = rd.getInput().getLineNumber(); 793 String location = getSourceLocation() != null ? getSourceLocation() : ""; 794 sb.append("\n <routeLocation") 795 .append(String.format( 796 " routeId=\"%s\" id=\"%s\" index=\"%s\" sourceLocation=\"%s\" sourceLineNumber=\"%s\"/>", 797 route.getRouteId(), id, 0, location, line)); 798 } 799 for (ManagedProcessorMBean processor : processors) { 800 // the step must belong to this route 801 if (route.getRouteId().equals(processor.getRouteId())) { 802 int line = processor.getSourceLineNumber() != null ? processor.getSourceLineNumber() : -1; 803 String location = processor.getSourceLocation() != null ? processor.getSourceLocation() : ""; 804 sb.append("\n <routeLocation") 805 .append(String.format( 806 " routeId=\"%s\" id=\"%s\" index=\"%s\" sourceLocation=\"%s\" sourceLineNumber=\"%s\"/>", 807 route.getRouteId(), processor.getProcessorId(), processor.getIndex(), location, line)); 808 } 809 } 810 } 811 sb.append("\n</routeLocations>"); 812 return sb.toString(); 813 } 814 815 @Override 816 public void reset(boolean includeProcessors) throws Exception { 817 reset(); 818 load.reset(); 819 thp.reset(); 820 821 // and now reset all processors for this route 822 if (includeProcessors) { 823 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 824 if (server != null) { 825 // get all the processor mbeans and sort them accordingly to their index 826 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 827 ObjectName query = ObjectName.getInstance( 828 jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*"); 829 QueryExp queryExp = Query.match(new AttributeValueExp("RouteId"), new StringValueExp(getRouteId())); 830 Set<ObjectName> names = server.queryNames(query, queryExp); 831 for (ObjectName name : names) { 832 server.invoke(name, "reset", null, null); 833 } 834 } 835 } 836 } 837 838 @Override 839 public void updateRouteFromXml(String xml) throws Exception { 840 // check whether this is allowed 841 if (!isUpdateRouteEnabled()) { 842 throw new IllegalAccessException("Updating route is not enabled"); 843 } 844 845 // convert to model from xml 846 ExtendedCamelContext ecc = context.getCamelContextExtension(); 847 InputStream is = context.getTypeConverter().convertTo(InputStream.class, xml); 848 RoutesDefinition routes = LwModelHelper.loadRoutesDefinition(is); 849 if (routes == null || routes.getRoutes().isEmpty()) { 850 return; 851 } 852 RouteDefinition def = routes.getRoutes().get(0); 853 854 // if the xml does not contain the route-id then we fix this by adding the actual route id 855 // this may be needed if the route-id was auto-generated, as the intend is to update this route 856 // and not add a new route, adding a new route, use the MBean operation on ManagedCamelContext instead. 857 if (ObjectHelper.isEmpty(def.getId())) { 858 def.setId(getRouteId()); 859 } else if (!def.getId().equals(getRouteId())) { 860 throw new IllegalArgumentException( 861 "Cannot update route from XML as routeIds does not match. routeId: " 862 + getRouteId() + ", routeId from XML: " + def.getId()); 863 } 864 865 LOG.debug("Updating route: {} from xml: {}", def.getId(), xml); 866 try { 867 // add will remove existing route first 868 ecc.getContextPlugin(Model.class).addRouteDefinition(def); 869 } catch (Exception e) { 870 // log the error as warn as the management api may be invoked remotely over JMX which does not propagate such exception 871 String msg = "Error updating route: " + def.getId() + " from xml: " + xml + " due: " + e.getMessage(); 872 LOG.warn(msg, e); 873 throw e; 874 } 875 } 876 877 @Override 878 public boolean isUpdateRouteEnabled() { 879 // check whether this is allowed 880 Boolean enabled = context.getManagementStrategy().getManagementAgent().getUpdateRouteEnabled(); 881 return enabled != null ? enabled : false; 882 } 883 884 @Override 885 public boolean isRemoteEndpoint() { 886 if (route.getEndpoint() != null) { 887 return route.getEndpoint().isRemote(); 888 } 889 return false; 890 } 891 892 @Override 893 public boolean equals(Object o) { 894 return this == o || o != null && getClass() == o.getClass() && route.equals(((ManagedRoute) o).route); 895 } 896 897 @Override 898 public int hashCode() { 899 return route.hashCode(); 900 } 901 902 private InflightRepository.InflightExchange getOldestInflightEntry() { 903 return getContext().getInflightRepository().oldest(getRouteId()); 904 } 905 906 @Override 907 public Long getOldestInflightDuration() { 908 InflightRepository.InflightExchange oldest = getOldestInflightEntry(); 909 if (oldest == null) { 910 return null; 911 } else { 912 return oldest.getDuration(); 913 } 914 } 915 916 @Override 917 public String getOldestInflightExchangeId() { 918 InflightRepository.InflightExchange oldest = getOldestInflightEntry(); 919 if (oldest == null) { 920 return null; 921 } else { 922 return oldest.getExchange().getExchangeId(); 923 } 924 } 925 926 @Override 927 public Boolean getHasRouteController() { 928 return route.getRouteController() != null; 929 } 930 931 @Override 932 public RouteError getLastError() { 933 org.apache.camel.spi.RouteError error = route.getLastError(); 934 if (error == null) { 935 return null; 936 } else { 937 return new RouteError() { 938 @Override 939 public Phase getPhase() { 940 if (error.getPhase() != null) { 941 switch (error.getPhase()) { 942 case START: 943 return Phase.START; 944 case STOP: 945 return Phase.STOP; 946 case SUSPEND: 947 return Phase.SUSPEND; 948 case RESUME: 949 return Phase.RESUME; 950 case SHUTDOWN: 951 return Phase.SHUTDOWN; 952 case REMOVE: 953 return Phase.REMOVE; 954 default: 955 throw new IllegalStateException(); 956 } 957 } 958 return null; 959 } 960 961 @Override 962 public Throwable getException() { 963 return error.getException(); 964 } 965 966 @Override 967 public Date getDate() { 968 return error.getDate(); 969 } 970 }; 971 } 972 } 973 974 @Override 975 public Collection<String> processorIds() throws Exception { 976 List<String> ids = new ArrayList<>(); 977 978 MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer(); 979 if (server != null) { 980 String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : ""; 981 // gather all the processors for this CamelContext, which requires JMX 982 ObjectName query = ObjectName 983 .getInstance(jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*"); 984 Set<ObjectName> names = server.queryNames(query, null); 985 for (ObjectName on : names) { 986 ManagedProcessorMBean processor 987 = context.getManagementStrategy().getManagementAgent().newProxyClient(on, ManagedProcessorMBean.class); 988 // the processor must belong to this route 989 if (getRouteId().equals(processor.getRouteId())) { 990 ids.add(processor.getProcessorId()); 991 } 992 } 993 } 994 995 return ids; 996 } 997 998 private Integer getInflightExchanges() { 999 return (int) super.getExchangesInflight(); 1000 } 1001 1002 /** 1003 * Used for sorting the processor mbeans accordingly to their index. 1004 */ 1005 private static final class OrderProcessorMBeans implements Comparator<ManagedProcessorMBean>, Serializable { 1006 1007 @Override 1008 public int compare(ManagedProcessorMBean o1, ManagedProcessorMBean o2) { 1009 return o1.getIndex().compareTo(o2.getIndex()); 1010 } 1011 } 1012}