mirror of
https://github.com/Athou/commafeed.git
synced 2026-03-21 21:37:29 +00:00
use codahale metrics library instead of our own
This commit is contained in:
@@ -1,179 +0,0 @@
|
||||
package com.commafeed.backend;
|
||||
|
||||
import javax.ejb.Singleton;
|
||||
import javax.interceptor.AroundInvoke;
|
||||
import javax.interceptor.InvocationContext;
|
||||
import javax.persistence.EntityManager;
|
||||
import javax.persistence.PersistenceContext;
|
||||
|
||||
import org.hibernate.Session;
|
||||
import org.hibernate.SessionFactory;
|
||||
import org.hibernate.stat.Statistics;
|
||||
|
||||
@Singleton
|
||||
public class MetricsBean {
|
||||
|
||||
@PersistenceContext
|
||||
EntityManager em;
|
||||
|
||||
private Metric lastMinute = new Metric();
|
||||
private Metric thisMinute = new Metric();
|
||||
|
||||
private Metric lastHour = new Metric();
|
||||
private Metric thisHour = new Metric();
|
||||
|
||||
private long minuteTimestamp;
|
||||
private long hourTimestamp;
|
||||
|
||||
@AroundInvoke
|
||||
private Object roll(InvocationContext context) throws Exception {
|
||||
long now = System.currentTimeMillis();
|
||||
if (now - minuteTimestamp > 60000) {
|
||||
lastMinute = thisMinute;
|
||||
thisMinute = new Metric();
|
||||
minuteTimestamp = now;
|
||||
|
||||
}
|
||||
|
||||
if (now - hourTimestamp > 60000 * 60) {
|
||||
lastHour = thisHour;
|
||||
thisHour = new Metric();
|
||||
hourTimestamp = now;
|
||||
|
||||
}
|
||||
return context.proceed();
|
||||
}
|
||||
|
||||
public void feedRefreshed() {
|
||||
thisMinute.feedsRefreshed++;
|
||||
thisHour.feedsRefreshed++;
|
||||
}
|
||||
|
||||
public void feedUpdated() {
|
||||
thisHour.feedsUpdated++;
|
||||
thisMinute.feedsUpdated++;
|
||||
|
||||
}
|
||||
|
||||
public void entryInserted() {
|
||||
|
||||
thisHour.entriesInserted++;
|
||||
thisMinute.entriesInserted++;
|
||||
|
||||
}
|
||||
|
||||
public void entryCacheHit() {
|
||||
thisHour.entryCacheHit++;
|
||||
thisMinute.entryCacheHit++;
|
||||
}
|
||||
|
||||
public void entryCacheMiss() {
|
||||
thisHour.entryCacheMiss++;
|
||||
thisMinute.entryCacheMiss++;
|
||||
}
|
||||
|
||||
public void pushReceived(int feedCount) {
|
||||
|
||||
thisHour.pushNotificationsReceived++;
|
||||
thisMinute.pushNotificationsReceived++;
|
||||
|
||||
thisHour.pushFeedsQueued += feedCount;
|
||||
thisMinute.pushFeedsQueued += feedCount;
|
||||
}
|
||||
|
||||
public void threadWaited() {
|
||||
thisHour.threadWaited++;
|
||||
thisMinute.threadWaited++;
|
||||
}
|
||||
|
||||
public Metric getLastMinute() {
|
||||
return lastMinute;
|
||||
}
|
||||
|
||||
public Metric getLastHour() {
|
||||
return lastHour;
|
||||
}
|
||||
|
||||
public String getCacheStats() {
|
||||
Session session = em.unwrap(Session.class);
|
||||
SessionFactory sessionFactory = session.getSessionFactory();
|
||||
Statistics statistics = sessionFactory.getStatistics();
|
||||
return statistics.toString();
|
||||
}
|
||||
|
||||
public static class Metric {
|
||||
private int feedsRefreshed;
|
||||
private int feedsUpdated;
|
||||
private int entriesInserted;
|
||||
private int threadWaited;
|
||||
private int pushNotificationsReceived;
|
||||
private int pushFeedsQueued;
|
||||
private int entryCacheHit;
|
||||
private int entryCacheMiss;
|
||||
|
||||
public int getFeedsRefreshed() {
|
||||
return feedsRefreshed;
|
||||
}
|
||||
|
||||
public void setFeedsRefreshed(int feedsRefreshed) {
|
||||
this.feedsRefreshed = feedsRefreshed;
|
||||
}
|
||||
|
||||
public int getFeedsUpdated() {
|
||||
return feedsUpdated;
|
||||
}
|
||||
|
||||
public void setFeedsUpdated(int feedsUpdated) {
|
||||
this.feedsUpdated = feedsUpdated;
|
||||
}
|
||||
|
||||
public int getEntriesInserted() {
|
||||
return entriesInserted;
|
||||
}
|
||||
|
||||
public void setEntriesInserted(int entriesInserted) {
|
||||
this.entriesInserted = entriesInserted;
|
||||
}
|
||||
|
||||
public int getThreadWaited() {
|
||||
return threadWaited;
|
||||
}
|
||||
|
||||
public void setThreadWaited(int threadWaited) {
|
||||
this.threadWaited = threadWaited;
|
||||
}
|
||||
|
||||
public int getPushNotificationsReceived() {
|
||||
return pushNotificationsReceived;
|
||||
}
|
||||
|
||||
public void setPushNotificationsReceived(int pushNotificationsReceived) {
|
||||
this.pushNotificationsReceived = pushNotificationsReceived;
|
||||
}
|
||||
|
||||
public int getPushFeedsQueued() {
|
||||
return pushFeedsQueued;
|
||||
}
|
||||
|
||||
public void setPushFeedsQueued(int pushFeedsQueued) {
|
||||
this.pushFeedsQueued = pushFeedsQueued;
|
||||
}
|
||||
|
||||
public int getEntryCacheHit() {
|
||||
return entryCacheHit;
|
||||
}
|
||||
|
||||
public void setEntryCacheHit(int entryCacheHit) {
|
||||
this.entryCacheHit = entryCacheHit;
|
||||
}
|
||||
|
||||
public int getEntryCacheMiss() {
|
||||
return entryCacheMiss;
|
||||
}
|
||||
|
||||
public void setEntryCacheMiss(int entryCacheMiss) {
|
||||
this.entryCacheMiss = entryCacheMiss;
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import javax.persistence.EntityManager;
|
||||
import javax.persistence.PersistenceContext;
|
||||
|
||||
import com.commafeed.backend.services.ApplicationSettingsService;
|
||||
import com.commafeed.backend.services.DatabaseCleaningService;
|
||||
|
||||
/**
|
||||
* Contains all scheduled tasks
|
||||
@@ -21,7 +22,7 @@ public class ScheduledTasks {
|
||||
ApplicationSettingsService applicationSettingsService;
|
||||
|
||||
@Inject
|
||||
DatabaseCleaner cleaner;
|
||||
DatabaseCleaningService cleaner;
|
||||
|
||||
@PersistenceContext
|
||||
EntityManager em;
|
||||
|
||||
@@ -7,6 +7,9 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import com.codahale.metrics.Gauge;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
/**
|
||||
* Wraps a {@link ThreadPoolExecutor} instance. Blocks when queue is full instead of rejecting the task. Allow priority queueing by using
|
||||
* {@link Task} instead of {@link Runnable}
|
||||
@@ -19,7 +22,7 @@ public class FeedRefreshExecutor {
|
||||
private ThreadPoolExecutor pool;
|
||||
private LinkedBlockingDeque<Runnable> queue;
|
||||
|
||||
public FeedRefreshExecutor(final String poolName, int threads, int queueCapacity) {
|
||||
public FeedRefreshExecutor(final String poolName, int threads, int queueCapacity, MetricRegistry metrics) {
|
||||
log.info("Creating pool {} with {} threads", poolName, threads);
|
||||
this.poolName = poolName;
|
||||
pool = new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, queue = new LinkedBlockingDeque<Runnable>(queueCapacity) {
|
||||
@@ -51,20 +54,26 @@ public class FeedRefreshExecutor {
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
metrics.register(MetricRegistry.name(getClass(), poolName, "active"), new Gauge<Integer>() {
|
||||
@Override
|
||||
public Integer getValue() {
|
||||
return pool.getActiveCount();
|
||||
}
|
||||
});
|
||||
|
||||
metrics.register(MetricRegistry.name(getClass(), poolName, "pending"), new Gauge<Integer>() {
|
||||
@Override
|
||||
public Integer getValue() {
|
||||
return queue.size();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public void execute(Task task) {
|
||||
pool.execute(task);
|
||||
}
|
||||
|
||||
public int getQueueSize() {
|
||||
return queue.size();
|
||||
}
|
||||
|
||||
public int getActiveCount() {
|
||||
return pool.getActiveCount();
|
||||
}
|
||||
|
||||
public static interface Task extends Runnable {
|
||||
boolean isUrgent();
|
||||
}
|
||||
|
||||
@@ -17,7 +17,8 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.codec.digest.DigestUtils;
|
||||
import org.apache.commons.lang.time.DateUtils;
|
||||
|
||||
import com.commafeed.backend.MetricsBean;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import com.commafeed.backend.dao.FeedDAO;
|
||||
import com.commafeed.backend.model.Feed;
|
||||
import com.commafeed.backend.services.ApplicationSettingsService;
|
||||
@@ -41,7 +42,7 @@ public class FeedRefreshTaskGiver {
|
||||
ApplicationSettingsService applicationSettingsService;
|
||||
|
||||
@Inject
|
||||
MetricsBean metricsBean;
|
||||
MetricRegistry metrics;
|
||||
|
||||
@Inject
|
||||
FeedRefreshWorker worker;
|
||||
@@ -54,10 +55,15 @@ public class FeedRefreshTaskGiver {
|
||||
|
||||
private ExecutorService executor;
|
||||
|
||||
private Meter feedRefreshed;
|
||||
private Meter threadWaited;
|
||||
|
||||
@PostConstruct
|
||||
public void init() {
|
||||
backgroundThreads = applicationSettingsService.get().getBackgroundThreads();
|
||||
executor = Executors.newFixedThreadPool(1);
|
||||
feedRefreshed = metrics.meter(MetricRegistry.name(getClass(), "feedRefreshed"));
|
||||
threadWaited = metrics.meter(MetricRegistry.name(getClass(), "threadWaited"));
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
@@ -88,11 +94,11 @@ public class FeedRefreshTaskGiver {
|
||||
try {
|
||||
FeedRefreshContext context = take();
|
||||
if (context != null) {
|
||||
metricsBean.feedRefreshed();
|
||||
feedRefreshed.mark();
|
||||
worker.updateFeed(context);
|
||||
} else {
|
||||
log.debug("nothing to do, sleeping for 15s");
|
||||
metricsBean.threadWaited();
|
||||
threadWaited.mark();
|
||||
try {
|
||||
Thread.sleep(15000);
|
||||
} catch (InterruptedException e) {
|
||||
|
||||
@@ -19,7 +19,8 @@ import org.apache.commons.collections.CollectionUtils;
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.apache.commons.lang3.time.DateUtils;
|
||||
|
||||
import com.commafeed.backend.MetricsBean;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import com.commafeed.backend.cache.CacheService;
|
||||
import com.commafeed.backend.dao.FeedDAO;
|
||||
import com.commafeed.backend.dao.FeedEntryDAO;
|
||||
@@ -57,7 +58,7 @@ public class FeedRefreshUpdater {
|
||||
ApplicationSettingsService applicationSettingsService;
|
||||
|
||||
@Inject
|
||||
MetricsBean metricsBean;
|
||||
MetricRegistry metrics;
|
||||
|
||||
@Inject
|
||||
FeedSubscriptionDAO feedSubscriptionDAO;
|
||||
@@ -71,12 +72,22 @@ public class FeedRefreshUpdater {
|
||||
private FeedRefreshExecutor pool;
|
||||
private Striped<Lock> locks;
|
||||
|
||||
private Meter entryCacheMiss;
|
||||
private Meter entryCacheHit;
|
||||
private Meter feedUpdated;
|
||||
private Meter entryInserted;
|
||||
|
||||
@PostConstruct
|
||||
public void init() {
|
||||
ApplicationSettings settings = applicationSettingsService.get();
|
||||
int threads = Math.max(settings.getDatabaseUpdateThreads(), 1);
|
||||
pool = new FeedRefreshExecutor("feed-refresh-updater", threads, Math.min(50 * threads, 1000));
|
||||
pool = new FeedRefreshExecutor("feed-refresh-updater", threads, Math.min(50 * threads, 1000), metrics);
|
||||
locks = Striped.lazyWeakLock(threads * 100000);
|
||||
|
||||
entryCacheMiss = metrics.meter(MetricRegistry.name(getClass(), "entryCacheMiss"));
|
||||
entryCacheHit = metrics.meter(MetricRegistry.name(getClass(), "entryCacheHit"));
|
||||
feedUpdated = metrics.meter(MetricRegistry.name(getClass(), "feedUpdated"));
|
||||
entryInserted = metrics.meter(MetricRegistry.name(getClass(), "entryInserted"));
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
@@ -116,10 +127,10 @@ public class FeedRefreshUpdater {
|
||||
subscriptions = feedSubscriptionDAO.findByFeed(feed);
|
||||
}
|
||||
ok &= addEntry(feed, entry, subscriptions);
|
||||
metricsBean.entryCacheMiss();
|
||||
entryCacheMiss.mark();
|
||||
} else {
|
||||
log.debug("cache hit for {}", entry.getUrl());
|
||||
metricsBean.entryCacheHit();
|
||||
entryCacheHit.mark();
|
||||
}
|
||||
|
||||
currentEntries.add(cacheKey);
|
||||
@@ -147,7 +158,7 @@ public class FeedRefreshUpdater {
|
||||
// requeue asap
|
||||
feed.setDisabledUntil(new Date(0));
|
||||
}
|
||||
metricsBean.feedUpdated();
|
||||
feedUpdated.mark();
|
||||
taskGiver.giveBack(feed);
|
||||
}
|
||||
|
||||
@@ -180,7 +191,7 @@ public class FeedRefreshUpdater {
|
||||
if (locked1 && locked2) {
|
||||
boolean inserted = feedUpdateService.addEntry(feed, entry);
|
||||
if (inserted) {
|
||||
metricsBean.entryInserted();
|
||||
entryInserted.mark();
|
||||
}
|
||||
success = true;
|
||||
} else {
|
||||
@@ -213,13 +224,4 @@ public class FeedRefreshUpdater {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public int getQueueSize() {
|
||||
return pool.getQueueSize();
|
||||
}
|
||||
|
||||
public int getActiveCount() {
|
||||
return pool.getActiveCount();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.codec.digest.DigestUtils;
|
||||
import org.apache.commons.lang.time.DateUtils;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import com.commafeed.backend.HttpGetter.NotModifiedException;
|
||||
import com.commafeed.backend.feeds.FeedRefreshExecutor.Task;
|
||||
import com.commafeed.backend.model.ApplicationSettings;
|
||||
@@ -37,6 +38,9 @@ public class FeedRefreshWorker {
|
||||
@Inject
|
||||
FeedRefreshTaskGiver taskGiver;
|
||||
|
||||
@Inject
|
||||
MetricRegistry metrics;
|
||||
|
||||
@Inject
|
||||
ApplicationSettingsService applicationSettingsService;
|
||||
|
||||
@@ -46,7 +50,7 @@ public class FeedRefreshWorker {
|
||||
private void init() {
|
||||
ApplicationSettings settings = applicationSettingsService.get();
|
||||
int threads = settings.getBackgroundThreads();
|
||||
pool = new FeedRefreshExecutor("feed-refresh-worker", threads, Math.min(20 * threads, 1000));
|
||||
pool = new FeedRefreshExecutor("feed-refresh-worker", threads, Math.min(20 * threads, 1000), metrics);
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
@@ -58,14 +62,6 @@ public class FeedRefreshWorker {
|
||||
pool.execute(new FeedTask(context));
|
||||
}
|
||||
|
||||
public int getQueueSize() {
|
||||
return pool.getQueueSize();
|
||||
}
|
||||
|
||||
public int getActiveCount() {
|
||||
return pool.getActiveCount();
|
||||
}
|
||||
|
||||
private class FeedTask implements Task {
|
||||
|
||||
private FeedRefreshContext context;
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
package com.commafeed.backend.metrics;
|
||||
|
||||
import javax.enterprise.context.ApplicationScoped;
|
||||
import javax.enterprise.inject.Produces;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
@ApplicationScoped
|
||||
public class MetricRegistryProducer {
|
||||
|
||||
private final MetricRegistry registry = new MetricRegistry();
|
||||
|
||||
@Produces
|
||||
public MetricRegistry produceMetricsRegistry() {
|
||||
return registry;
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
package com.commafeed.backend.feeds;
|
||||
package com.commafeed.backend.opml;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
@@ -1,4 +1,4 @@
|
||||
package com.commafeed.backend.feeds;
|
||||
package com.commafeed.backend.opml;
|
||||
|
||||
import java.io.StringReader;
|
||||
import java.util.List;
|
||||
@@ -15,6 +15,7 @@ import org.apache.commons.lang.StringUtils;
|
||||
|
||||
import com.commafeed.backend.cache.CacheService;
|
||||
import com.commafeed.backend.dao.FeedCategoryDAO;
|
||||
import com.commafeed.backend.feeds.FeedUtils;
|
||||
import com.commafeed.backend.model.FeedCategory;
|
||||
import com.commafeed.backend.model.User;
|
||||
import com.commafeed.backend.services.FeedSubscriptionService;
|
||||
@@ -1,4 +1,4 @@
|
||||
package com.commafeed.backend;
|
||||
package com.commafeed.backend.services;
|
||||
|
||||
import java.util.Calendar;
|
||||
import java.util.Date;
|
||||
@@ -16,14 +16,13 @@ import com.commafeed.backend.dao.FeedEntryStatusDAO;
|
||||
import com.commafeed.backend.dao.FeedSubscriptionDAO;
|
||||
import com.commafeed.backend.model.Feed;
|
||||
import com.commafeed.backend.model.FeedSubscription;
|
||||
import com.commafeed.backend.services.ApplicationSettingsService;
|
||||
|
||||
/**
|
||||
* Contains utility methods for cleaning the database
|
||||
*
|
||||
*/
|
||||
@Slf4j
|
||||
public class DatabaseCleaner {
|
||||
public class DatabaseCleaningService {
|
||||
|
||||
@Inject
|
||||
FeedDAO feedDAO;
|
||||
@@ -1,4 +1,4 @@
|
||||
package com.commafeed.backend;
|
||||
package com.commafeed.backend.startup;
|
||||
|
||||
import java.sql.Connection;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package com.commafeed.backend;
|
||||
package com.commafeed.backend.startup;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.io.InputStreamReader;
|
||||
Reference in New Issue
Block a user