downloader/downloader/main.py

@@ -62,18 +62,6 @@ segment_time_skew = prom.Histogram(
 	buckets=[-10, -1, -0.5, -0.1, -0.01, -0.001, 0, 0.001, 0.01, 0.1, 0.5, 1, 10],
 )
 
-segment_time_skew_non_zero_sum = prom.Gauge(
-	"segment_time_skew_non_zero_sum",
-	"",
-	["channel", "quality", "worker"],
-)
-
-segment_time_skew_non_zero_count = prom.Counter(
-	"segment_time_skew_non_zero_count",
-	"",
-	["channel", "quality", "worker"],
-)
-
 
 class TimedOutError(Exception):
 	pass
@@ -378,15 +366,12 @@ class StreamWorker(object):
 	def _run(self):
 		first = True
 		suspicious_skew = False
-		history = []
-		HISTORY_SIZE = 10
 		while not self.stopping.is_set():
 
 			self.logger.debug("Getting media playlist {}".format(self.url))
 			try:
 				with soft_hard_timeout(self.logger, "getting media playlist", self.FETCH_TIMEOUTS, self.trigger_new_worker):
-					playlist_time = datetime.datetime.utcnow()
-					raw_playlist, playlist = self.manager.provider.get_media_playlist(self.url, session=self.session)
+					playlist = self.manager.provider.get_media_playlist(self.url, session=self.session)
 			except Exception as e:
 				self.logger.warning("Failed to fetch media playlist {}".format(self.url), exc_info=True)
 				self.trigger_new_worker()
@@ -402,11 +387,8 @@ class StreamWorker(object):
 			# We successfully got the playlist at least once
 			first = False
 
-			history = [(playlist_time, raw_playlist)] + history[:HISTORY_SIZE]
-
 			# Start any new segment getters
 			date = None # tracks date in case some segment doesn't include it
-			prev_segment = None
 			for segment in playlist.segments:
 				if segment.ad_reason:
 					self.logger.info("Ignoring ad segment: {}".format(segment.ad_reason))
@@ -429,11 +411,7 @@ class StreamWorker(object):
 						#      but only used if no other source is available.
 						skew = (date - new_date).total_seconds()
 						segment_time_skew.labels(self.manager.channel, self.quality, f"{id(self):x}").observe(skew)
-						if skew != 0:
-							segment_time_skew_non_zero_sum.labels(self.manager.channel, self.quality, f"{id(self):x}").inc(skew)
-							segment_time_skew_non_zero_count.labels(self.manager.channel, self.quality, f"{id(self):x}").inc()
 						if abs(skew) > self.MAX_SEGMENT_TIME_SKEW and not suspicious_skew:
-							self.logger.warning(f"Suspicious skew of {skew}, triggering new worker: Expected {date} after {prev_segment}, got {new_date} for {segment}")
 							self.trigger_new_worker()
 							suspicious_skew = True
 							suspicious_skew_count.labels(self.manager.channel, self.quality).inc()
@@ -451,12 +429,10 @@ class StreamWorker(object):
 						date,
 						suspicious_skew,
 						self.map_cache,
-						history,
 					)
 					gevent.spawn(self.getters[segment.uri].run)
 				if date is not None:
 					date += datetime.timedelta(seconds=segment.duration)
-				prev_segment = segment
 
 			# Clean up any old segment getters.
 			# Note use of list() to make a copy to avoid modification-during-iteration
@@ -508,7 +484,7 @@ class SegmentGetter(object):
 	# or so, to be paranoid we set it to considerably longer than that.
 	GIVE_UP_TIMEOUT = 20 * 60
 
-	def __init__(self, parent_logger, session, base_dir, channel, quality, segment, date, suspect, map_cache, history):
+	def __init__(self, parent_logger, session, base_dir, channel, quality, segment, date, suspect, map_cache):
 		self.logger = parent_logger.getChild("SegmentGetter@{:x}".format(id(self)))
 		self.base_dir = base_dir
 		self.channel = channel
@@ -522,7 +498,6 @@ class SegmentGetter(object):
 		self.done = gevent.event.Event() # set when file exists or we give up
 		# Our parent's connection pool, but we'll replace it if there's any issues
 		self.session = session
-		self.history = history
 
 	def run(self):
 		try:
@@ -682,19 +657,6 @@ class SegmentGetter(object):
 			stat = latest_segment.labels(channel=self.channel, quality=self.quality)
 			timestamp = (self.date - datetime.datetime(1970, 1, 1)).total_seconds()
 			stat.set(max(stat._value.get(), timestamp)) # NOTE: not thread-safe but is gevent-safe
-			self.write_history(full_path)
-
-	def write_history(self, segment_path):
-		segment_path = os.path.relpath(segment_path, self.base_dir)
-		history_path = os.path.join(self.base_dir, "playlist-debug", segment_path)
-		try:
-			os.makedirs(history_path)
-		except FileExistsError:
-			pass
-		for n, (timestamp, playlist) in enumerate(self.history):
-			filename = "{}_{}".format(n, timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f"))
-			path = os.path.join(history_path, filename)
-			common.atomic_write(path, playlist)
 
 
 def parse_channel(channel):

downloader/downloader/providers.py

@@ -29,8 +29,7 @@ class Provider:
 			session = InstrumentedSession()
 		resp = session.get(uri, metric_name='get_media_playlist')
 		resp.raise_for_status()
-		playlist = resp.text
-		return playlist, hls_playlist.load(playlist, base_uri=resp.url)
+		return hls_playlist.load(resp.text, base_uri=resp.url)
 
 
 class URLProvider(Provider):