diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 9f0c4c19113b..b2223244f863 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -84,6 +84,7 @@ https://github.com/elastic/beats/compare/v6.4.0...6.x[Check the HEAD diff]
 
 *Affecting all Beats*
 
+- Added time-based log rotation. {pull}8349[8349]
 - Add backoff on error support to redis output. {pull}7781[7781]
 - Allow for cloud-id to specify a custom port. This makes cloud-id work in ECE contexts. {pull}7887[7887]
 - Add support to grow or shrink an existing spool file between restarts. {pull}7859[7859]
diff --git a/auditbeat/auditbeat.reference.yml b/auditbeat/auditbeat.reference.yml
index b31493eb02f3..2da8d120ae45 100644
--- a/auditbeat/auditbeat.reference.yml
+++ b/auditbeat/auditbeat.reference.yml
@@ -530,11 +530,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1089,6 +1089,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/filebeat/filebeat.reference.yml b/filebeat/filebeat.reference.yml
index f22078b937ae..67a73a7d2f46 100644
--- a/filebeat/filebeat.reference.yml
+++ b/filebeat/filebeat.reference.yml
@@ -1203,11 +1203,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1762,6 +1762,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/heartbeat/heartbeat.reference.yml b/heartbeat/heartbeat.reference.yml
index d94143f42847..aaf18c1f423d 100644
--- a/heartbeat/heartbeat.reference.yml
+++ b/heartbeat/heartbeat.reference.yml
@@ -650,11 +650,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1209,6 +1209,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/libbeat/_meta/config.reference.yml b/libbeat/_meta/config.reference.yml
index 61066c4ac808..be17b5884b62 100644
--- a/libbeat/_meta/config.reference.yml
+++ b/libbeat/_meta/config.reference.yml
@@ -423,11 +423,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -982,6 +982,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/libbeat/common/file/interval_rotator.go b/libbeat/common/file/interval_rotator.go
new file mode 100644
index 000000000000..b531acca2373
--- /dev/null
+++ b/libbeat/common/file/interval_rotator.go
@@ -0,0 +1,194 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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 file
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+	"strconv"
+	"time"
+)
+
+type intervalRotator struct {
+	interval    time.Duration
+	lastRotate  time.Time
+	fileFormat  string
+	clock       clock
+	weekly      bool
+	arbitrary   bool
+	newInterval func(lastTime time.Time, currentTime time.Time) bool
+}
+
+type clock interface {
+	Now() time.Time
+}
+
+type realClock struct{}
+
+func (realClock) Now() time.Time {
+	return time.Now()
+}
+
+func newIntervalRotator(interval time.Duration) (*intervalRotator, error) {
+	if interval == 0 {
+		return nil, nil
+	}
+	if interval < time.Second && interval != 0 {
+		return nil, errors.New("the minimum time interval for log rotation is 1 second")
+	}
+
+	ir := &intervalRotator{interval: (interval / time.Second) * time.Second} // drop fractional seconds
+	ir.initialize()
+	return ir, nil
+}
+
+func (r *intervalRotator) initialize() error {
+	r.clock = realClock{}
+
+	switch r.interval {
+	case time.Second:
+		r.fileFormat = "2006-01-02-15-04-05"
+		r.newInterval = newSecond
+	case time.Minute:
+		r.fileFormat = "2006-01-02-15-04"
+		r.newInterval = newMinute
+	case time.Hour:
+		r.fileFormat = "2006-01-02-15"
+		r.newInterval = newHour
+	case 24 * time.Hour: // calendar day
+		r.fileFormat = "2006-01-02"
+		r.newInterval = newDay
+	case 7 * 24 * time.Hour: // calendar week
+		r.fileFormat = ""
+		r.newInterval = newWeek
+		r.weekly = true
+	case 30 * 24 * time.Hour: // calendar month
+		r.fileFormat = "2006-01"
+		r.newInterval = newMonth
+	case 365 * 24 * time.Hour: // calendar year
+		r.fileFormat = "2006"
+		r.newInterval = newYear
+	default:
+		r.arbitrary = true
+		r.fileFormat = "2006-01-02-15-04-05"
+		r.newInterval = func(lastTime time.Time, currentTime time.Time) bool {
+			lastInterval := lastTime.Unix() / (int64(r.interval) / int64(time.Second))
+			currentInterval := currentTime.Unix() / (int64(r.interval) / int64(time.Second))
+			return lastInterval != currentInterval
+		}
+	}
+	return nil
+}
+
+func (r *intervalRotator) LogPrefix(filename string, modTime time.Time) string {
+	var t time.Time
+	if r.lastRotate.IsZero() {
+		t = modTime
+	} else {
+		t = r.lastRotate
+	}
+
+	if r.weekly {
+		y, w := t.ISOWeek()
+		return fmt.Sprintf("%s-%04d-%02d-", filename, y, w)
+	}
+	if r.arbitrary {
+		intervalNumber := t.Unix() / (int64(r.interval) / int64(time.Second))
+		intervalStart := time.Unix(0, intervalNumber*int64(r.interval))
+		return fmt.Sprintf("%s-%s-", filename, intervalStart.Format(r.fileFormat))
+	}
+	return fmt.Sprintf("%s-%s-", filename, t.Format(r.fileFormat))
+}
+
+func (r *intervalRotator) NewInterval() bool {
+	now := r.clock.Now()
+	newInterval := r.newInterval(r.lastRotate, now)
+	return newInterval
+}
+
+func (r *intervalRotator) Rotate() {
+	r.lastRotate = r.clock.Now()
+}
+
+func (r *intervalRotator) SortIntervalLogs(strings []string) {
+	sort.Slice(
+		strings,
+		func(i, j int) bool {
+			return OrderIntervalLogs(strings[i]) < OrderIntervalLogs(strings[j])
+		},
+	)
+}
+
+// OrderIntervalLogs, when given a log filename in the form [prefix]-[formattedDate]-n
+// returns the filename after zero-padding the trailing n so that foo-[date]-2 sorts
+// before foo-[date]-10.
+func OrderIntervalLogs(filename string) string {
+	index, i, err := IntervalLogIndex(filename)
+	if err == nil {
+		return filename[:i] + fmt.Sprintf("%020d", index)
+	}
+
+	return ""
+}
+
+// IntervalLogIndex returns n as int given a log filename in the form [prefix]-[formattedDate]-n
+func IntervalLogIndex(filename string) (uint64, int, error) {
+	i := len(filename) - 1
+	for ; i >= 0; i-- {
+		if '0' > filename[i] || filename[i] > '9' {
+			break
+		}
+	}
+	i++
+
+	s64 := filename[i:]
+	u64, err := strconv.ParseUint(s64, 10, 64)
+	return u64, i, err
+}
+
+func newSecond(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Second() != currentTime.Second() || newMinute(lastTime, currentTime)
+}
+
+func newMinute(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Minute() != currentTime.Minute() || newHour(lastTime, currentTime)
+}
+
+func newHour(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Hour() != currentTime.Hour() || newDay(lastTime, currentTime)
+}
+
+func newDay(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Day() != currentTime.Day() || newMonth(lastTime, currentTime)
+}
+
+func newWeek(lastTime time.Time, currentTime time.Time) bool {
+	lastYear, lastWeek := lastTime.ISOWeek()
+	currentYear, currentWeek := currentTime.ISOWeek()
+	return lastWeek != currentWeek ||
+		lastYear != currentYear
+}
+
+func newMonth(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Month() != currentTime.Month() || newYear(lastTime, currentTime)
+}
+
+func newYear(lastTime time.Time, currentTime time.Time) bool {
+	return lastTime.Year() != currentTime.Year()
+}
diff --git a/libbeat/common/file/interval_rotator_test.go b/libbeat/common/file/interval_rotator_test.go
new file mode 100644
index 000000000000..fda516d86271
--- /dev/null
+++ b/libbeat/common/file/interval_rotator_test.go
@@ -0,0 +1,277 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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 file
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSecondRotator(t *testing.T) {
+	a, err := newIntervalRotator(time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 31, 0, 0, 1, 100, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-00-01-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(900 * time.Millisecond)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-00-01-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(100 * time.Millisecond)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-00-02-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestMinuteRotator(t *testing.T) {
+	a, err := newIntervalRotator(time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 31, 0, 1, 1, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-01-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(58 * time.Second)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-01-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-02-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestHourlyRotator(t *testing.T) {
+	a, err := newIntervalRotator(time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 31, 1, 0, 1, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-01-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(58 * time.Minute)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-01-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Minute + 59*time.Second)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-02-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestDailyRotator(t *testing.T) {
+	a, err := newIntervalRotator(24 * time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 31, 0, 0, 0, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(23 * time.Hour)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Hour)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2019-01-01-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestWeeklyRotator(t *testing.T) {
+	a, err := newIntervalRotator(7 * 24 * time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Monday, 2018-Dec-31
+	clock := &testClock{time.Date(2018, 12, 31, 0, 0, 0, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2019-01-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	// Sunday, 2019-Jan-6
+	clock.time = clock.time.Add(6 * 24 * time.Hour)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2019-01-", a.LogPrefix("foo", time.Now()))
+
+	// Monday, 2019-Jan-7
+	clock.time = clock.time.Add(24 * time.Hour)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2019-02-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestMonthlyRotator(t *testing.T) {
+	a, err := newIntervalRotator(30 * 24 * time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 1, 0, 0, 0, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(30 * 24 * time.Hour)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(24 * time.Hour)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2019-01-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestYearlyRotator(t *testing.T) {
+	a, err := newIntervalRotator(365 * 24 * time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clock := &testClock{time.Date(2018, 12, 31, 0, 0, 0, 0, time.Local)}
+	a.clock = clock
+	a.Rotate()
+	assert.Equal(t, "foo-2018-", a.LogPrefix("foo", time.Now()))
+
+	n := a.NewInterval()
+	assert.False(t, n)
+
+	clock.time = clock.time.Add(23 * time.Hour)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Hour)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2019-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestArbitraryIntervalRotator(t *testing.T) {
+	a, err := newIntervalRotator(3 * time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Monday, 2018-Dec-31
+	clock := &testClock{time.Date(2018, 12, 31, 0, 0, 1, 0, time.Local)}
+	a.clock = clock
+	assert.Equal(t, "foo-2018-12-30-00-00-00-", a.LogPrefix("foo", time.Date(2018, 12, 30, 0, 0, 0, 0, time.Local)))
+	a.Rotate()
+	n := a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-00-00-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-00-00-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-00-03-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-00-03-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.False(t, n)
+	assert.Equal(t, "foo-2018-12-31-00-00-03-", a.LogPrefix("foo", time.Now()))
+
+	clock.time = clock.time.Add(time.Second)
+	n = a.NewInterval()
+	assert.True(t, n)
+	a.Rotate()
+	assert.Equal(t, "foo-2018-12-31-00-00-06-", a.LogPrefix("foo", time.Now()))
+}
+
+func TestIntervalIsTruncatedToSeconds(t *testing.T) {
+	a, err := newIntervalRotator(2345 * time.Millisecond)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, 2*time.Second, a.interval)
+}
+
+func TestZeroIntervalIsNil(t *testing.T) {
+	a, err := newIntervalRotator(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.True(t, a == nil)
+}
+
+type testClock struct {
+	time time.Time
+}
+
+func (t testClock) Now() time.Time {
+	return t.time
+}
diff --git a/libbeat/common/file/rotator.go b/libbeat/common/file/rotator.go
index 45446a9c4a34..bf83f24ea709 100644
--- a/libbeat/common/file/rotator.go
+++ b/libbeat/common/file/rotator.go
@@ -22,6 +22,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 )
@@ -37,6 +38,7 @@ const (
 	rotateReasonInitializing rotateReason = iota + 1
 	rotateReasonFileSize
 	rotateReasonManualTrigger
+	rotateReasonTimeInterval
 )
 
 func (rr rotateReason) String() string {
@@ -47,20 +49,25 @@ func (rr rotateReason) String() string {
 		return "file size"
 	case rotateReasonManualTrigger:
 		return "manual trigger"
+	case rotateReasonTimeInterval:
+		return "time interval"
 	default:
 		return "unknown"
 	}
 }
 
 // Rotator is a io.WriteCloser that automatically rotates the file it is
-// writing to when it reaches a maximum size. It also purges the oldest rotated
-// files when the maximum number of backups is reached.
+// writing to when it reaches a maximum size and optionally on a time interval
+// basis. It also purges the oldest rotated files when the maximum number of
+// backups is reached.
 type Rotator struct {
-	filename     string
-	maxSizeBytes uint
-	maxBackups   uint
-	permissions  os.FileMode
-	log          Logger // Optional Logger (may be nil).
+	filename        string
+	maxSizeBytes    uint
+	maxBackups      uint
+	permissions     os.FileMode
+	log             Logger // Optional Logger (may be nil).
+	interval        time.Duration
+	intervalRotator *intervalRotator // Optional, may be nil
 
 	file  *os.File
 	size  uint
@@ -108,6 +115,14 @@ func WithLogger(l Logger) RotatorOption {
 	}
 }
 
+// Interval sets the time interval for log rotation in addition to log
+// rotation by size. The default is 0 for disabled.
+func Interval(d time.Duration) RotatorOption {
+	return func(r *Rotator) {
+		r.interval = d
+	}
+}
+
 // NewFileRotator returns a new Rotator.
 func NewFileRotator(filename string, options ...RotatorOption) (*Rotator, error) {
 	r := &Rotator{
@@ -115,6 +130,7 @@ func NewFileRotator(filename string, options ...RotatorOption) (*Rotator, error)
 		maxSizeBytes: 10 * 1024 * 1024, // 10 MiB
 		maxBackups:   7,
 		permissions:  0600,
+		interval:     0,
 	}
 
 	for _, opt := range options {
@@ -130,6 +146,11 @@ func NewFileRotator(filename string, options ...RotatorOption) (*Rotator, error)
 	if r.permissions > os.ModePerm {
 		return nil, errors.Errorf("file rotator permissions mask of %o is invalid", r.permissions)
 	}
+	var err error
+	r.intervalRotator, err = newIntervalRotator(r.interval)
+	if err != nil {
+		return nil, err
+	}
 
 	if r.log != nil {
 		r.log.Debugw("Initialized file rotator",
@@ -137,6 +158,7 @@ func NewFileRotator(filename string, options ...RotatorOption) (*Rotator, error)
 			"max_size_bytes", r.maxSizeBytes,
 			"max_backups", r.maxBackups,
 			"permissions", r.permissions,
+			"interval", r.interval,
 		)
 	}
 
@@ -160,6 +182,13 @@ func (r *Rotator) Write(data []byte) (int, error) {
 		if err := r.openNew(); err != nil {
 			return 0, err
 		}
+	} else if r.intervalRotator != nil && r.intervalRotator.NewInterval() {
+		if err := r.rotate(rotateReasonTimeInterval); err != nil {
+			return 0, err
+		}
+		if err := r.openFile(); err != nil {
+			return 0, err
+		}
 	} else if r.size+dataLen > r.maxSizeBytes {
 		if err := r.rotate(rotateReasonFileSize); err != nil {
 			return 0, err
@@ -263,6 +292,43 @@ func (r *Rotator) closeFile() error {
 }
 
 func (r *Rotator) purgeOldBackups() error {
+	if r.intervalRotator != nil {
+		return r.purgeOldIntervalBackups()
+	}
+	return r.purgeOldSizedBackups()
+}
+
+func (r *Rotator) purgeOldIntervalBackups() error {
+	files, err := filepath.Glob(r.filename + "*")
+	if err != nil {
+		return errors.Wrap(err, "failed to list existing logs during rotation")
+	}
+
+	if len(files) > int(r.maxBackups) {
+
+		// sort log filenames numerically
+		r.intervalRotator.SortIntervalLogs(files)
+
+		for i := len(files) - int(r.maxBackups) - 1; i >= 0; i-- {
+			f := files[i]
+			_, err := os.Stat(f)
+			switch {
+			case err == nil:
+				if err = os.Remove(f); err != nil {
+					return errors.Wrapf(err, "failed to delete %v during rotation", f)
+				}
+			case os.IsNotExist(err):
+				return errors.Wrapf(err, "failed to delete non-existent %v during rotation", f)
+			default:
+				return errors.Wrapf(err, "failed on %v during rotation", f)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (r *Rotator) purgeOldSizedBackups() error {
 	for i := r.maxBackups; i < MaxBackupsLimit; i++ {
 		name := r.backupName(i + 1)
 
@@ -287,6 +353,59 @@ func (r *Rotator) rotate(reason rotateReason) error {
 		return errors.Wrap(err, "error file closing current file")
 	}
 
+	var err error
+	if r.intervalRotator != nil {
+		err = r.rotateByInterval(reason)
+	} else {
+		err = r.rotateBySize(reason)
+	}
+	if err != nil {
+		return errors.Wrap(err, "failed to rotate backups")
+	}
+
+	return r.purgeOldBackups()
+}
+
+func (r *Rotator) rotateByInterval(reason rotateReason) error {
+	fi, err := os.Stat(r.filename)
+	if os.IsNotExist(err) {
+		return nil
+	} else if err != nil {
+		return errors.Wrap(err, "failed to rotate backups")
+	}
+
+	logPrefix := r.intervalRotator.LogPrefix(r.filename, fi.ModTime())
+	files, err := filepath.Glob(logPrefix + "*")
+	if err != nil {
+		return errors.Wrap(err, "failed to list logs during rotation")
+	}
+
+	var targetFilename string
+	if len(files) == 0 {
+		targetFilename = logPrefix + "1"
+	} else {
+		r.intervalRotator.SortIntervalLogs(files)
+		lastLogIndex, _, err := IntervalLogIndex(files[len(files)-1])
+		if err != nil {
+			return errors.Wrap(err, "failed to locate last log index during rotation")
+		}
+		targetFilename = logPrefix + strconv.Itoa(int(lastLogIndex)+1)
+	}
+
+	if err := os.Rename(r.filename, targetFilename); err != nil {
+		return errors.Wrap(err, "failed to rotate backups")
+	}
+
+	if r.log != nil {
+		r.log.Debugw("Rotating file", "filename", r.filename, "reason", reason)
+	}
+
+	r.intervalRotator.Rotate()
+
+	return nil
+}
+
+func (r *Rotator) rotateBySize(reason rotateReason) error {
 	for i := r.maxBackups + 1; i > 0; i-- {
 		old := r.backupName(i - 1)
 		older := r.backupName(i)
@@ -309,6 +428,5 @@ func (r *Rotator) rotate(reason rotateReason) error {
 			}
 		}
 	}
-
-	return r.purgeOldBackups()
+	return nil
 }
diff --git a/libbeat/common/file/rotator_test.go b/libbeat/common/file/rotator_test.go
index 16cf6ada9749..07995e7e29ce 100644
--- a/libbeat/common/file/rotator_test.go
+++ b/libbeat/common/file/rotator_test.go
@@ -24,6 +24,7 @@ import (
 	"sort"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -99,6 +100,86 @@ func TestFileRotatorConcurrently(t *testing.T) {
 	wg.Wait()
 }
 
+func TestDailyRotation(t *testing.T) {
+	dir, err := ioutil.TempDir("", "daily_file_rotator")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	logname := "daily"
+	dateFormat := "2006-01-02"
+	today := time.Now().Format(dateFormat)
+	yesterday := time.Now().AddDate(0, 0, -1).Format(dateFormat)
+	twoDaysAgo := time.Now().AddDate(0, 0, -2).Format(dateFormat)
+
+	// seed directory with existing log files
+	files := []string{
+		logname + "-" + yesterday + "-1",
+		logname + "-" + yesterday + "-2",
+		logname + "-" + yesterday + "-3",
+		logname + "-" + yesterday + "-4",
+		logname + "-" + yesterday + "-5",
+		logname + "-" + yesterday + "-6",
+		logname + "-" + yesterday + "-7",
+		logname + "-" + yesterday + "-8",
+		logname + "-" + yesterday + "-9",
+		logname + "-" + yesterday + "-10",
+		logname + "-" + yesterday + "-11",
+		logname + "-" + yesterday + "-12",
+		logname + "-" + yesterday + "-13",
+		logname + "-" + twoDaysAgo + "-1",
+		logname + "-" + twoDaysAgo + "-2",
+		logname + "-" + twoDaysAgo + "-3",
+	}
+
+	for _, f := range files {
+		CreateFile(t, filepath.Join(dir, f))
+	}
+
+	maxSizeBytes := uint(500)
+	filename := filepath.Join(dir, logname)
+	r, err := file.NewFileRotator(filename, file.MaxBackups(2), file.Interval(24*time.Hour), file.MaxSizeBytes(maxSizeBytes))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	Rotate(t, r)
+
+	AssertDirContents(t, dir, logname+"-"+yesterday+"-12", logname+"-"+yesterday+"-13")
+
+	WriteMsg(t, r)
+
+	AssertDirContents(t, dir, logname+"-"+yesterday+"-12", logname+"-"+yesterday+"-13", logname)
+
+	Rotate(t, r)
+
+	AssertDirContents(t, dir, logname+"-"+yesterday+"-13", logname+"-"+today+"-1")
+
+	WriteMsg(t, r)
+
+	AssertDirContents(t, dir, logname+"-"+yesterday+"-13", logname+"-"+today+"-1", logname)
+
+	for i := 0; i < (int(maxSizeBytes)/len(logMessage))+1; i++ {
+		WriteMsg(t, r)
+	}
+
+	AssertDirContents(t, dir, logname+"-"+today+"-1", logname+"-"+today+"-2", logname)
+}
+
+func CreateFile(t *testing.T, filename string) {
+	t.Helper()
+	f, err := os.Create(filename)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = f.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func AssertDirContents(t *testing.T, dir string, files ...string) {
 	t.Helper()
 
diff --git a/libbeat/docs/loggingconfig.asciidoc b/libbeat/docs/loggingconfig.asciidoc
index f897321f4c92..db34c284a8da 100644
--- a/libbeat/docs/loggingconfig.asciidoc
+++ b/libbeat/docs/loggingconfig.asciidoc
@@ -155,6 +155,15 @@ Examples:
 * 0664: give read and write access to the file owner and members of the group
 associated with the file, as well as read access to all other users.
 
+[float]
+==== `logging.files.interval`
+
+Enable log file rotation on time intervals in addition to size-based rotation.
+Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+are boundary-aligned with minutes, hours, days, weeks, months, and years as
+reported by the local system clock. All other intervals are calculated from the
+unix epoch. Defaults to disabled.
+
 [float]
 ==== `logging.json`
 
diff --git a/libbeat/logp/config.go b/libbeat/logp/config.go
index dea2d2a44210..e028c9b690dd 100644
--- a/libbeat/logp/config.go
+++ b/libbeat/logp/config.go
@@ -17,6 +17,8 @@
 
 package logp
 
+import "time"
+
 // Config contains the configuration options for the logger. To create a Config
 // from a common.Config use logp/config.Build.
 type Config struct {
@@ -40,11 +42,12 @@ type Config struct {
 
 // FileConfig contains the configuration options for the file output.
 type FileConfig struct {
-	Path        string `config:"path"`
-	Name        string `config:"name"`
-	MaxSize     uint   `config:"rotateeverybytes" validate:"min=1"`
-	MaxBackups  uint   `config:"keepfiles" validate:"max=1024"`
-	Permissions uint32 `config:"permissions"`
+	Path        string        `config:"path"`
+	Name        string        `config:"name"`
+	MaxSize     uint          `config:"rotateeverybytes" validate:"min=1"`
+	MaxBackups  uint          `config:"keepfiles" validate:"max=1024"`
+	Permissions uint32        `config:"permissions"`
+	Interval    time.Duration `config:"interval"`
 }
 
 var defaultConfig = Config{
@@ -54,6 +57,7 @@ var defaultConfig = Config{
 		MaxSize:     10 * 1024 * 1024,
 		MaxBackups:  7,
 		Permissions: 0600,
+		Interval:    0,
 	},
 	addCaller: true,
 }
diff --git a/libbeat/logp/core.go b/libbeat/logp/core.go
index f117dc2ef731..00762c7fb599 100644
--- a/libbeat/logp/core.go
+++ b/libbeat/logp/core.go
@@ -195,6 +195,7 @@ func makeFileOutput(cfg Config) (zapcore.Core, error) {
 		file.MaxSizeBytes(cfg.Files.MaxSize),
 		file.MaxBackups(cfg.Files.MaxBackups),
 		file.Permissions(os.FileMode(cfg.Files.Permissions)),
+		file.Interval(cfg.Files.Interval),
 	)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create file rotator")
diff --git a/metricbeat/metricbeat.reference.yml b/metricbeat/metricbeat.reference.yml
index fc82566f257d..243641be4c8e 100644
--- a/metricbeat/metricbeat.reference.yml
+++ b/metricbeat/metricbeat.reference.yml
@@ -1097,11 +1097,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1656,6 +1656,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/packetbeat/packetbeat.reference.yml b/packetbeat/packetbeat.reference.yml
index a6175cc6d3ca..e41211552b1c 100644
--- a/packetbeat/packetbeat.reference.yml
+++ b/packetbeat/packetbeat.reference.yml
@@ -904,11 +904,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1463,6 +1463,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false
 
diff --git a/winlogbeat/winlogbeat.reference.yml b/winlogbeat/winlogbeat.reference.yml
index b686b5746d62..263181953aaf 100644
--- a/winlogbeat/winlogbeat.reference.yml
+++ b/winlogbeat/winlogbeat.reference.yml
@@ -452,11 +452,11 @@ output.elasticsearch:
   # and retry until all events are published.  Set max_retries to a value less
   # than 0 to retry until all events are published. The default is 3.
   #max_retries: 3
-  
+
   # The maximum number of events to bulk in a single Logstash request. The
   # default is 2048.
   #bulk_max_size: 2048
-  
+
   # The number of seconds to wait for responses from the Logstash server before
   # timing out. The default is 30s.
   #timeout: 30s
@@ -1011,6 +1011,13 @@ logging.files:
   # Must be a valid Unix-style file permissions mask expressed in octal notation.
   #permissions: 0600
 
+  # Enable log file rotation on time intervals in addition to size-based rotation.
+  # Intervals must be at least 1s. Values of 1m, 1h, 24h, 7*24h, 30*24h, and 365*24h
+  # are boundary-aligned with minutes, hours, days, weeks, months, and years as
+  # reported by the local system clock. All other intervals are calculated from the
+  # unix epoch. Defaults to disabled.
+  #interval: 0
+
 # Set to true to log messages in json format.
 #logging.json: false