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}