| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.karaf.audit; |
| |
| import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; |
| import org.apache.karaf.audit.layout.GelfLayout; |
| import org.apache.karaf.audit.layout.Rfc3164Layout; |
| import org.apache.karaf.audit.layout.Rfc5424Layout; |
| import org.apache.karaf.audit.layout.SimpleLayout; |
| import org.apache.karaf.audit.logger.FileEventLogger; |
| import org.apache.karaf.audit.logger.JulEventLogger; |
| import org.apache.karaf.audit.logger.UdpEventLogger; |
| import org.apache.karaf.util.tracker.BaseActivator; |
| import org.apache.karaf.util.tracker.annotation.Managed; |
| import org.apache.karaf.util.tracker.annotation.RequireService; |
| import org.apache.karaf.util.tracker.annotation.Services; |
| import org.osgi.framework.Filter; |
| import org.osgi.framework.FrameworkUtil; |
| import org.osgi.framework.InvalidSyntaxException; |
| import org.osgi.service.cm.ManagedService; |
| import org.osgi.service.event.Event; |
| import org.osgi.service.event.EventAdmin; |
| import org.osgi.service.event.EventConstants; |
| import org.osgi.service.event.EventHandler; |
| |
| import javax.security.auth.Subject; |
| import java.io.IOException; |
| import java.lang.reflect.Constructor; |
| import java.util.AbstractMap; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.Dictionary; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.Hashtable; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Map; |
| import java.util.NoSuchElementException; |
| import java.util.Set; |
| import java.util.TimeZone; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| |
| @Services(requires = @RequireService(EventAdmin.class)) |
| @Managed("org.apache.karaf.audit") |
| public class Activator extends BaseActivator implements ManagedService { |
| |
| public static final String FILTER = "filter"; |
| public static final String QUEUE_TYPE = "queue.type"; |
| public static final String QUEUE_SIZE = "queue.size"; |
| public static final String RUNNER_IDLE_TIMEOUT = "runner.idle-timeout"; |
| public static final String RUNNER_FLUSH_TIMEOUT = "runner.flush-timeout"; |
| public static final String FILE_PREFIX = "file."; |
| public static final String FILE_LAYOUT = FILE_PREFIX + "layout"; |
| public static final String FILE_ENABLED = FILE_PREFIX + "enabled"; |
| public static final String FILE_TARGET = FILE_PREFIX + "target"; |
| public static final String FILE_ENCODING = FILE_PREFIX + "encoding"; |
| public static final String FILE_POLICY = FILE_PREFIX + "policy"; |
| public static final String FILE_FILES = FILE_PREFIX + "files"; |
| public static final String FILE_COMPRESS = FILE_PREFIX + "compress"; |
| public static final String UDP_PREFIX = "udp."; |
| public static final String UDP_LAYOUT = UDP_PREFIX + "layout"; |
| public static final String UDP_ENABLED = UDP_PREFIX + "enabled"; |
| public static final String UDP_HOST = UDP_PREFIX + "host"; |
| public static final String UDP_PORT = UDP_PREFIX + "port"; |
| public static final String UDP_ENCODING = UDP_PREFIX + "encoding"; |
| public static final String TCP_PREFIX = "tcp."; |
| public static final String TCP_LAYOUT = TCP_PREFIX + "layout"; |
| public static final String TCP_ENABLED = TCP_PREFIX + "enabled"; |
| public static final String TCP_HOST = TCP_PREFIX + "host"; |
| public static final String TCP_PORT = TCP_PREFIX + "port"; |
| public static final String TCP_ENCODING = TCP_PREFIX + "encoding"; |
| public static final String JUL_PREFIX = "jul."; |
| public static final String JUL_LAYOUT = JUL_PREFIX + "layout"; |
| public static final String JUL_ENABLED = JUL_PREFIX + "enabled"; |
| public static final String JUL_LOGGER = JUL_PREFIX + "logger"; |
| public static final String JUL_LEVEL = JUL_PREFIX + "level"; |
| public static final String TOPICS = "topics"; |
| |
| private static final EventImpl STOP_EVENT = new EventImpl(new Event("stop", Collections.emptyMap())); |
| |
| |
| private BlockingQueue<EventImpl> queue; |
| private volatile Thread runner; |
| private List<EventLogger> eventLoggers; |
| private Filter filter; |
| |
| @Override |
| protected void doStart() throws Exception { |
| super.doStart(); |
| queue = createQueue(); |
| eventLoggers = createLoggers(); |
| filter = createFilter(); |
| final Dictionary<String, Object> props = new Hashtable<>(); |
| props.put(EventConstants.EVENT_TOPIC, getTopics()); |
| register(EventHandler.class, this::handleEvent, props); |
| if (!queue.isEmpty()) { |
| startRunner(); |
| } |
| } |
| |
| private String[] getTopics() { |
| return getString(TOPICS, "*").split("\\s*,\\s*"); |
| } |
| |
| private Filter createFilter() throws InvalidSyntaxException { |
| String str = getString(FILTER, null); |
| return str != null ? FrameworkUtil.createFilter(str) : null; |
| } |
| |
| @SuppressWarnings("unchecked") |
| private BlockingQueue<EventImpl> createQueue() throws Exception { |
| String type = getString(QUEUE_TYPE, null); |
| int size = getInt(QUEUE_SIZE, 1024); |
| if ("ArrayBlockingQueue".equals(type)) { |
| return new ArrayBlockingQueue<>(size); |
| } else if ("DisruptorBlockingQueue".equals(type)) { |
| return new DisruptorBlockingQueue(size); |
| } else if (type != null) { |
| logger.warn("Unknown queue type: " + type + ""); |
| } |
| try { |
| return new DisruptorBlockingQueue(size); |
| } catch (NoClassDefFoundError t) { |
| return new ArrayBlockingQueue<>(size); |
| } |
| } |
| |
| private List<EventLogger> createLoggers() throws Exception { |
| try { |
| List<EventLogger> loggers = new ArrayList<>(); |
| if (getBoolean(FILE_ENABLED, true)) { |
| String path = getString(FILE_TARGET, System.getProperty("karaf.data") + "/log/audit.txt"); |
| String encoding = getString(FILE_ENCODING, "UTF-8"); |
| String policy = getString(FILE_POLICY, "size(8mb)"); |
| int files = getInt(FILE_FILES, 32); |
| boolean compress = getBoolean(FILE_COMPRESS, true); |
| EventLayout layout = createLayout(getString(FILE_LAYOUT, FILE_LAYOUT)); |
| loggers.add(new FileEventLogger(path, encoding, policy, files, compress, this, layout, TimeZone.getDefault())); |
| } |
| if (getBoolean(UDP_ENABLED, false)) { |
| String host = getString(UDP_HOST, "localhost"); |
| int port = getInt(UDP_PORT, 514); |
| String encoding = getString(UDP_ENCODING, "UTF-8"); |
| EventLayout layout = createLayout(getString(UDP_LAYOUT, UDP_LAYOUT)); |
| loggers.add(new UdpEventLogger(host, port, encoding, layout)); |
| } |
| if (getBoolean(TCP_ENABLED, false)) { |
| String host = getString(TCP_HOST, "localhost"); |
| int port = getInt(TCP_PORT, 0); |
| String encoding = getString(TCP_ENCODING, "UTF-8"); |
| EventLayout layout = createLayout(getString(TCP_LAYOUT, TCP_LAYOUT)); |
| loggers.add(new UdpEventLogger(host, port, encoding, layout)); |
| } |
| if (getBoolean(JUL_ENABLED, false)) { |
| String logger = getString(Activator.JUL_LOGGER, "audit"); |
| String level = getString(Activator.JUL_LEVEL, "info"); |
| EventLayout layout = createLayout(getString(JUL_LAYOUT, JUL_LAYOUT)); |
| loggers.add(new JulEventLogger(logger, level, layout)); |
| } |
| return loggers; |
| } catch (IOException e) { |
| throw new Exception("Error creating audit logger", e); |
| } |
| } |
| |
| private EventLayout createLayout(String prefix) { |
| String type = getString(prefix + ".type", "simple"); |
| switch (type) { |
| case "simple": |
| return new SimpleLayout(); |
| case "rfc3164": |
| return new Rfc3164Layout(getInt(prefix + ".facility", 16), |
| getInt(prefix + ".priority", 5), |
| getInt(prefix + ".enterprise", Rfc5424Layout.DEFAULT_ENTERPRISE_NUMBER), |
| TimeZone.getDefault(), |
| Locale.ENGLISH); |
| case "rfc5424": |
| return new Rfc5424Layout(getInt(prefix + ".facility", 16), |
| getInt(prefix + ".priority", 5), |
| getInt(prefix + ".enterprise", Rfc5424Layout.DEFAULT_ENTERPRISE_NUMBER), |
| TimeZone.getDefault()); |
| case "gelf": |
| return new GelfLayout(); |
| default: |
| try { |
| return createCustomLayout(type); |
| } catch (Exception e) { |
| logger.error("Error creating layout: " + type + ". Using a simple layout.", e); |
| return new SimpleLayout(); |
| } |
| } |
| } |
| |
| private EventLayout createCustomLayout(String type) throws ClassNotFoundException, InstantiationException, IllegalAccessException, java.lang.reflect.InvocationTargetException { |
| Class<?> clazz = Class.forName(type); |
| Constructor<?> cnsMap = null; |
| Constructor<?> cnsDef = null; |
| Object layout = null; |
| try { |
| cnsMap = clazz.getConstructor(Map.class); |
| } catch (NoSuchMethodException e) { |
| // ignore |
| } |
| try { |
| cnsDef = clazz.getConstructor(); |
| } catch (NoSuchMethodException e) { |
| // ignore |
| } |
| if (cnsMap != null) { |
| Map<String, Object> params = new HashMap<>(); |
| for (Enumeration<String> e = getConfiguration().keys(); e.hasMoreElements(); ) { |
| String key = e.nextElement(); |
| Object val = getConfiguration().get(key); |
| params.put(key, val); |
| } |
| layout = cnsMap.newInstance(params); |
| } else if (cnsDef != null) { |
| layout = cnsDef.newInstance(); |
| } else { |
| throw new IllegalArgumentException("Unable to find a supported constructor"); |
| } |
| if (layout instanceof EventLayout) { |
| return (EventLayout) layout; |
| } else { |
| throw new IllegalArgumentException("The built layout does not implement " + EventLayout.class.getName()); |
| } |
| } |
| |
| @Override |
| protected void doStop() { |
| Thread runner = this.runner; |
| if (runner != null && runner.isAlive()) { |
| try { |
| queue.add(STOP_EVENT); |
| runner.join(5000); |
| if (runner.isAlive()) { |
| runner.interrupt(); |
| } |
| } catch (InterruptedException e) { |
| logger.debug("Error waiting for audit runner buffer stop"); |
| } |
| } |
| List<EventLogger> eventLoggers = this.eventLoggers; |
| if (eventLoggers != null) { |
| for (EventLogger eventLogger : eventLoggers) { |
| try { |
| eventLogger.close(); |
| } catch (IOException e) { |
| logger.debug("Error closing audit logger", e); |
| } |
| } |
| this.eventLoggers = null; |
| } |
| super.doStop(); |
| } |
| |
| private void handleEvent(Event event) { |
| try { |
| EventImpl ev = new EventImpl(event); |
| if (filter == null || filter.matches(ev.getFilterMap())) { |
| queue.put(new EventImpl(event)); |
| startRunner(); |
| } |
| } catch (InterruptedException e) { |
| logger.debug("Interrupted while putting event in queue", e); |
| } |
| } |
| |
| private void startRunner() { |
| if (eventLoggers != null && !eventLoggers.isEmpty() && runner == null) { |
| synchronized (this) { |
| if (runner == null) { |
| runner = new Thread(this::consume, "audit-logger"); |
| runner.start(); |
| } |
| } |
| } |
| } |
| |
| private void consume() { |
| long maxIdle = getLong(RUNNER_IDLE_TIMEOUT, TimeUnit.MINUTES.toMillis(1)); |
| long flushDelay = getLong(RUNNER_FLUSH_TIMEOUT, TimeUnit.MILLISECONDS.toMillis(100)); |
| try { |
| List<EventLogger> eventLoggers = this.eventLoggers; |
| BlockingQueue<EventImpl> queue = this.queue; |
| EventImpl event; |
| while ((event = queue.poll(maxIdle, TimeUnit.MILLISECONDS)) != null) { |
| if (event == STOP_EVENT) { |
| return; |
| } |
| for (EventLogger eventLogger : eventLoggers) { |
| eventLogger.write(event); |
| } |
| if (flushDelay > 0) { |
| while ((event = queue.poll(flushDelay, TimeUnit.MILLISECONDS)) != null) { |
| if (event == STOP_EVENT) { |
| return; |
| } |
| for (EventLogger eventLogger : eventLoggers) { |
| eventLogger.write(event); |
| } |
| } |
| } |
| for (EventLogger eventLogger : eventLoggers) { |
| eventLogger.flush(); |
| } |
| } |
| } catch (Throwable e) { |
| logger.warn("Error writing audit log", e); |
| } finally { |
| runner = null; |
| } |
| } |
| |
| static class EventImpl implements org.apache.karaf.audit.Event { |
| private final Event event; |
| private final long timestamp; |
| private final String type; |
| private final String subtype; |
| |
| EventImpl(Event event) { |
| this.event = event; |
| this.timestamp = _timestamp(); |
| this.type = _type(); |
| this.subtype = _subtype(); |
| } |
| |
| @Override |
| public long timestamp() { |
| return timestamp; |
| } |
| |
| private long _timestamp() { |
| Long l = (Long) event.getProperty("timestamp"); |
| return l != null ? l : System.currentTimeMillis(); |
| } |
| |
| @Override |
| public Subject subject() { |
| return (Subject) event.getProperty("subject"); |
| } |
| |
| @Override |
| public String type() { |
| return type; |
| } |
| |
| private String _type() { |
| switch (event.getTopic()) { |
| case "org/apache/karaf/shell/console/EXECUTED": |
| return TYPE_SHELL; |
| case "org/osgi/service/log/LogEntry/LOG_ERROR": |
| case "org/osgi/service/log/LogEntry/LOG_WARNING": |
| case "org/osgi/service/log/LogEntry/LOG_INFO": |
| case "org/osgi/service/log/LogEntry/LOG_DEBUG": |
| case "org/osgi/service/log/LogEntry/LOG_OTHER": |
| return TYPE_LOG; |
| case "org/osgi/framework/ServiceEvent/REGISTERED": |
| case "org/osgi/framework/ServiceEvent/MODIFIED": |
| case "org/osgi/framework/ServiceEvent/UNREGISTERING": |
| return TYPE_SERVICE; |
| case "org/osgi/framework/BundleEvent/INSTALLED": |
| case "org/osgi/framework/BundleEvent/STARTED": |
| case "org/osgi/framework/BundleEvent/STOPPED": |
| case "org/osgi/framework/BundleEvent/UPDATED": |
| case "org/osgi/framework/BundleEvent/UNINSTALLED": |
| case "org/osgi/framework/BundleEvent/RESOLVED": |
| case "org/osgi/framework/BundleEvent/UNRESOLVED": |
| case "org/osgi/framework/BundleEvent/STARTING": |
| case "org/osgi/framework/BundleEvent/STOPPING": |
| return TYPE_BUNDLE; |
| case "org/apache/karaf/login/ATTEMPT": |
| case "org/apache/karaf/login/SUCCESS": |
| case "org/apache/karaf/login/FAILURE": |
| case "org/apache/karaf/login/LOGOUT": |
| return TYPE_LOGIN; |
| case "javax/management/MBeanServer/CREATEMBEAN": |
| case "javax/management/MBeanServer/REGISTERMBEAN": |
| case "javax/management/MBeanServer/UNREGISTERMBEAN": |
| case "javax/management/MBeanServer/GETOBJECTINSTANCE": |
| case "javax/management/MBeanServer/QUERYMBEANS": |
| case "javax/management/MBeanServer/ISREGISTERED": |
| case "javax/management/MBeanServer/GETMBEANCOUNT": |
| case "javax/management/MBeanServer/GETATTRIBUTE": |
| case "javax/management/MBeanServer/GETATTRIBUTES": |
| case "javax/management/MBeanServer/SETATTRIBUTE": |
| case "javax/management/MBeanServer/SETATTRIBUTES": |
| case "javax/management/MBeanServer/INVOKE": |
| case "javax/management/MBeanServer/GETDEFAULTDOMAIN": |
| case "javax/management/MBeanServer/GETDOMAINS": |
| case "javax/management/MBeanServer/ADDNOTIFICATIONLISTENER": |
| case "javax/management/MBeanServer/GETMBEANINFO": |
| case "javax/management/MBeanServer/ISINSTANCEOF": |
| case "javax/management/MBeanServer/INSTANTIATE": |
| case "javax/management/MBeanServer/DESERIALIZE": |
| case "javax/management/MBeanServer/GETCLASSLOADERFOR": |
| case "javax/management/MBeanServer/GETCLASSLOADER": |
| return TYPE_JMX; |
| case "org/osgi/framework/FrameworkEvent/STARTED": |
| case "org/osgi/framework/FrameworkEvent/ERROR": |
| case "org/osgi/framework/FrameworkEvent/PACKAGES_REFRESHED": |
| case "org/osgi/framework/FrameworkEvent/STARTLEVEL_CHANGED": |
| case "org/osgi/framework/FrameworkEvent/WARNING": |
| case "org/osgi/framework/FrameworkEvent/INFO": |
| case "org/osgi/framework/FrameworkEvent/STOPPED": |
| case "org/osgi/framework/FrameworkEvent/STOPPED_UPDATE": |
| case "org/osgi/framework/FrameworkEvent/STOPPED_BOOTCLASSPATH_MODIFIED": |
| case "org/osgi/framework/FrameworkEvent/WAIT_TIMEDOUT": |
| return TYPE_FRAMEWORK; |
| case "org/osgi/service/web/DEPLOYING": |
| case "org/osgi/service/web/DEPLOYED": |
| case "org/osgi/service/web/UNDEPLOYING": |
| case "org/osgi/service/web/UNDEPLOYED": |
| return TYPE_WEB; |
| case "org/apache/karaf/features/repositories/ADDED": |
| case "org/apache/karaf/features/repositories/REMOVED": |
| return TYPE_REPOSITORIES; |
| case "org/apache/karaf/features/features/INSTALLED": |
| case "org/apache/karaf/features/features/UNINSTALLED": |
| return TYPE_FEATURES; |
| case "org/osgi/service/blueprint/container/CREATING": |
| case "org/osgi/service/blueprint/container/CREATED": |
| case "org/osgi/service/blueprint/container/DESTROYING": |
| case "org/osgi/service/blueprint/container/DESTROYED": |
| case "org/osgi/service/blueprint/container/FAILURE": |
| case "org/osgi/service/blueprint/container/GRACE_PERIOD": |
| case "org/osgi/service/blueprint/container/WAITING": |
| return TYPE_BLUEPRINT; |
| default: |
| return TYPE_UNKNOWN; |
| } |
| } |
| |
| @Override |
| public String subtype() { |
| return subtype; |
| } |
| |
| private String _subtype() { |
| String topic = event.getTopic(); |
| String subtype = topic.substring(topic.lastIndexOf('/') + 1).toLowerCase(Locale.ENGLISH); |
| if (subtype.startsWith("log_")) { |
| subtype = subtype.substring("log_".length()); |
| } |
| return subtype; |
| } |
| |
| @Override |
| public Iterable<String> keys() { |
| String[] keys = event.getPropertyNames(); |
| Arrays.sort(keys); |
| return () -> new Iterator<String>() { |
| String next; |
| int index = -1; |
| @Override |
| public boolean hasNext() { |
| if (next != null) { |
| return true; |
| } |
| while (++index < keys.length) { |
| switch (keys[index]) { |
| case "timestamp": |
| case "event.topics": |
| case "subject": |
| case "type": |
| case "subtype": |
| break; |
| default: |
| next = keys[index]; |
| return true; |
| } |
| } |
| return false; |
| } |
| @Override |
| public String next() { |
| if (!hasNext()) { |
| throw new NoSuchElementException(); |
| } |
| String str = next; |
| next = null; |
| return str; |
| } |
| }; |
| } |
| |
| @Override |
| public Object getProperty(String key) { |
| return event.getProperty(key); |
| } |
| |
| Map<String, Object> getFilterMap() { |
| return new AbstractMap<String, Object>() { |
| @Override |
| public Set<Entry<String, Object>> entrySet() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public Object get(Object key) { |
| String s = key.toString(); |
| switch (s) { |
| case "timestamp": |
| return timestamp(); |
| case "type": |
| return type(); |
| case "subtype": |
| return subtype(); |
| case "subject": |
| return subject(); |
| default: |
| return event.getProperty(s); |
| } |
| } |
| }; |
| } |
| |
| } |
| |
| } |
