Migrate from usage of channels to waitgroups on app/analytics.go (#16929)

* Migrate from usage of channels to waitgroups on app/analytics.go

* Grouping error handling to simplify the code

* Using errgroup.Group instead of sync.WaitGroup

* Returning correctly the errors

* Update app/analytics.go

Co-authored-by: Miguel de la Cruz <mgdelacroix@gmail.com>

* Renaming g with g2 for clarity

Co-authored-by: Miguel de la Cruz <mgdelacroix@gmail.com>

app/analytics.go

@@ -6,9 +6,10 @@ package app
 import (
 	"net/http"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/mattermost/mattermost-server/v5/mlog"
 	"github.com/mattermost/mattermost-server/v5/model"
-	"github.com/mattermost/mattermost-server/v5/store"
 )
 
 const (
@@ -43,115 +44,105 @@ func (a *App) GetAnalytics(name string, teamID string) (model.AnalyticsRows, *mo
 		rows[9] = &model.AnalyticsRow{Name: "monthly_active_users", Value: 0}
 		rows[10] = &model.AnalyticsRow{Name: "inactive_user_count", Value: 0}
 
-		openChan := make(chan store.StoreResult, 1)
+		var g errgroup.Group
-		privateChan := make(chan store.StoreResult, 1)
+		var openChannelsCount int64
-		go func() {
+		g.Go(func() error {
-			count, err2 := a.Srv().Store.Channel().AnalyticsTypeCount(teamID, model.CHANNEL_OPEN)
+			var err error
-			openChan <- store.StoreResult{Data: count, NErr: err2}
+			if openChannelsCount, err = a.Srv().Store.Channel().AnalyticsTypeCount(teamID, model.CHANNEL_OPEN); err != nil {
-			close(openChan)
+				return model.NewAppError("GetAnalytics", "app.channel.analytics_type_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-		}()
+			}
-		go func() {
+			return nil
-			count, err2 := a.Srv().Store.Channel().AnalyticsTypeCount(teamID, model.CHANNEL_PRIVATE)
+		})
-			privateChan <- store.StoreResult{Data: count, NErr: err2}
-			close(privateChan)
-		}()
 
-		var userChan chan store.StoreResult
+		var privateChannelsCount int64
-		var userInactiveChan chan store.StoreResult
+		g.Go(func() error {
+			var err error
+			if privateChannelsCount, err = a.Srv().Store.Channel().AnalyticsTypeCount(teamID, model.CHANNEL_PRIVATE); err != nil {
+				return model.NewAppError("GetAnalytics", "app.channel.analytics_type_count.app_error", nil, err.Error(), http.StatusInternalServerError)
+			}
+			return nil
+		})
+
+		var usersCount int64
+		var inactiveUsersCount int64
 		if teamID == "" {
-			userInactiveChan = make(chan store.StoreResult, 1)
+			g.Go(func() error {
-			go func() {
+				var err error
-				count, err2 := a.Srv().Store.User().AnalyticsGetInactiveUsersCount()
+				if inactiveUsersCount, err = a.Srv().Store.User().AnalyticsGetInactiveUsersCount(); err != nil {
-				userInactiveChan <- store.StoreResult{Data: count, NErr: err2}
+					return model.NewAppError("GetAnalytics", "app.user.analytics_get_inactive_users_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-				close(userInactiveChan)
+				}
-			}()
+				return nil
+			})
 		} else {
-			userChan = make(chan store.StoreResult, 1)
+			g.Go(func() error {
-			go func() {
+				var err error
-				count, err2 := a.Srv().Store.User().Count(model.UserCountOptions{TeamId: teamID})
+				if usersCount, err = a.Srv().Store.User().Count(model.UserCountOptions{TeamId: teamID}); err != nil {
-				userChan <- store.StoreResult{Data: count, NErr: err2}
+					return model.NewAppError("GetAnalytics", "app.user.get_total_users_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-				close(userChan)
+				}
-			}()
+				return nil
+			})
 		}
 
-		var postChan chan store.StoreResult
+		var postsCount int64
 		if !skipIntensiveQueries {
-			postChan = make(chan store.StoreResult, 1)
+			g.Go(func() error {
-			go func() {
+				var err error
-				count, err2 := a.Srv().Store.Post().AnalyticsPostCount(teamID, false, false)
+				if postsCount, err = a.Srv().Store.Post().AnalyticsPostCount(teamID, false, false); err != nil {
-				postChan <- store.StoreResult{Data: count, NErr: err2}
+					return model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-				close(postChan)
+				}
-			}()
+				return nil
+			})
 		}
 
-		teamCountChan := make(chan store.StoreResult, 1)
+		var teamsCount int64
-		go func() {
+		g.Go(func() error {
-			teamCount, err2 := a.Srv().Store.Team().AnalyticsTeamCount(false)
+			var err error
-			teamCountChan <- store.StoreResult{Data: teamCount, NErr: err2}
+			if teamsCount, err = a.Srv().Store.Team().AnalyticsTeamCount(false); err != nil {
-			close(teamCountChan)
+				return model.NewAppError("GetAnalytics", "app.team.analytics_team_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-		}()
+			}
+			return nil
+		})
 
-		dailyActiveChan := make(chan store.StoreResult, 1)
+		var dailyActiveUsersCount int64
-		go func() {
+		g.Go(func() error {
-			dailyActive, err2 := a.Srv().Store.User().AnalyticsActiveCount(DayMilliseconds, model.UserCountOptions{IncludeBotAccounts: false, IncludeDeleted: false})
+			var err error
-			dailyActiveChan <- store.StoreResult{Data: dailyActive, NErr: err2}
+			if dailyActiveUsersCount, err = a.Srv().Store.User().AnalyticsActiveCount(DayMilliseconds, model.UserCountOptions{IncludeBotAccounts: false, IncludeDeleted: false}); err != nil {
-			close(dailyActiveChan)
+				return model.NewAppError("GetAnalytics", "app.user.analytics_daily_active_users.app_error", nil, err.Error(), http.StatusInternalServerError)
-		}()
+			}
+			return nil
+		})
 
-		monthlyActiveChan := make(chan store.StoreResult, 1)
+		var monthlyActiveUsersCount int64
-		go func() {
+		g.Go(func() error {
-			monthlyActive, err2 := a.Srv().Store.User().AnalyticsActiveCount(MonthMilliseconds, model.UserCountOptions{IncludeBotAccounts: false, IncludeDeleted: false})
+			var err error
-			monthlyActiveChan <- store.StoreResult{Data: monthlyActive, NErr: err2}
+			if monthlyActiveUsersCount, err = a.Srv().Store.User().AnalyticsActiveCount(MonthMilliseconds, model.UserCountOptions{IncludeBotAccounts: false, IncludeDeleted: false}); err != nil {
-			close(monthlyActiveChan)
+				return model.NewAppError("GetAnalytics", "app.user.analytics_daily_active_users.app_error", nil, err.Error(), http.StatusInternalServerError)
-		}()
+			}
+			return nil
+		})
 
-		r := <-openChan
+		if err := g.Wait(); err != nil {
-		if r.NErr != nil {
+			return nil, err.(*model.AppError)
-			return nil, model.NewAppError("GetAnalytics", "app.channel.analytics_type_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
 		}
-		rows[0].Value = float64(r.Data.(int64))
 
-		r = <-privateChan
+		rows[0].Value = float64(openChannelsCount)
-		if r.NErr != nil {
+		rows[1].Value = float64(privateChannelsCount)
-			return nil, model.NewAppError("GetAnalytics", "app.channel.analytics_type_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[1].Value = float64(r.Data.(int64))
 
-		if postChan == nil {
+		if skipIntensiveQueries {
 			rows[2].Value = -1
 		} else {
-			r = <-postChan
+			rows[2].Value = float64(postsCount)
-			if r.NErr != nil {
-				return nil, model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-			}
-			rows[2].Value = float64(r.Data.(int64))
 		}
 
-		if userChan == nil {
+		if teamID == "" {
 			rows[3].Value = float64(systemUserCount)
+			rows[10].Value = float64(inactiveUsersCount)
 		} else {
-			r = <-userChan
-			if r.NErr != nil {
-				return nil, model.NewAppError("GetAnalytics", "app.user.get_total_users_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-			}
-			rows[3].Value = float64(r.Data.(int64))
-		}
-
-		if userInactiveChan == nil {
 			rows[10].Value = -1
-		} else {
+			rows[3].Value = float64(usersCount)
-			r = <-userInactiveChan
-			if r.NErr != nil {
-				return nil, model.NewAppError("GetAnalytics", "app.user.analytics_get_inactive_users_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-			}
-			rows[10].Value = float64(r.Data.(int64))
 		}
 
-		r = <-teamCountChan
+		rows[4].Value = float64(teamsCount)
-		if r.NErr != nil {
-			return nil, model.NewAppError("GetAnalytics", "app.team.analytics_team_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[4].Value = float64(r.Data.(int64))
 
 		// If in HA mode then aggregate all the stats
 		if a.Cluster() != nil && *a.Config().ClusterSettings.Enable {
@@ -180,17 +171,8 @@ func (a *App) GetAnalytics(name string, teamID string) (model.AnalyticsRows, *mo
 			rows[7].Value = float64(a.Srv().Store.TotalReadDbConnections())
 		}
 
-		r = <-dailyActiveChan
+		rows[8].Value = float64(dailyActiveUsersCount)
-		if r.NErr != nil {
+		rows[9].Value = float64(monthlyActiveUsersCount)
-			return nil, model.NewAppError("GetAnalytics", "app.user.analytics_daily_active_users.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[8].Value = float64(r.Data.(int64))
-
-		r = <-monthlyActiveChan
-		if r.NErr != nil {
-			return nil, model.NewAppError("GetAnalytics", "app.user.analytics_daily_active_users.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[9].Value = float64(r.Data.(int64))
 
 		return rows, nil
 	} else if name == "bot_post_counts_day" {
@@ -244,96 +226,80 @@ func (a *App) GetAnalytics(name string, teamID string) (model.AnalyticsRows, *mo
 		rows[4] = &model.AnalyticsRow{Name: "command_count", Value: 0}
 		rows[5] = &model.AnalyticsRow{Name: "session_count", Value: 0}
 
-		iHookChan := make(chan store.StoreResult, 1)
+		var g2 errgroup.Group
-		go func() {
-			c, err2 := a.Srv().Store.Webhook().AnalyticsIncomingCount(teamID)
-			iHookChan <- store.StoreResult{Data: c, NErr: err2}
-			close(iHookChan)
-		}()
 
-		oHookChan := make(chan store.StoreResult, 1)
+		var incomingWebhookCount int64
-		go func() {
+		g2.Go(func() error {
-			c, err2 := a.Srv().Store.Webhook().AnalyticsOutgoingCount(teamID)
+			var err error
-			oHookChan <- store.StoreResult{Data: c, NErr: err2}
+			if incomingWebhookCount, err = a.Srv().Store.Webhook().AnalyticsIncomingCount(teamID); err != nil {
-			close(oHookChan)
+				return model.NewAppError("GetAnalytics", "app.webhooks.analytics_incoming_count.app_error", nil, err.Error(), http.StatusInternalServerError)
-		}()
-
-		commandChan := make(chan store.StoreResult, 1)
-		go func() {
-			c, nErr := a.Srv().Store.Command().AnalyticsCommandCount(teamID)
-			commandChan <- store.StoreResult{Data: c, NErr: nErr}
-			close(commandChan)
-		}()
-
-		sessionChan := make(chan store.StoreResult, 1)
-		go func() {
-			count, err2 := a.Srv().Store.Session().AnalyticsSessionCount()
-			sessionChan <- store.StoreResult{Data: count, NErr: err2}
-			close(sessionChan)
-		}()
-
-		var fileChan chan store.StoreResult
-		var hashtagChan chan store.StoreResult
-
-		if !skipIntensiveQueries {
-			fileChan = make(chan store.StoreResult, 1)
-			go func() {
-				count, err2 := a.Srv().Store.Post().AnalyticsPostCount(teamID, true, false)
-				fileChan <- store.StoreResult{Data: count, NErr: err2}
-				close(fileChan)
-			}()
-
-			hashtagChan = make(chan store.StoreResult, 1)
-			go func() {
-				count, err2 := a.Srv().Store.Post().AnalyticsPostCount(teamID, false, true)
-				hashtagChan <- store.StoreResult{Data: count, NErr: err2}
-				close(hashtagChan)
-			}()
-		}
-
-		if fileChan == nil {
-			rows[0].Value = -1
-		} else {
-			r := <-fileChan
-			if r.NErr != nil {
-				return nil, model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
 			}
-			rows[0].Value = float64(r.Data.(int64))
+			return nil
+		})
+
+		var outgoingWebhookCount int64
+		g2.Go(func() error {
+			var err error
+			if outgoingWebhookCount, err = a.Srv().Store.Webhook().AnalyticsOutgoingCount(teamID); err != nil {
+				return model.NewAppError("GetAnalytics", "app.webhooks.analytics_outgoing_count.app_error", nil, err.Error(), http.StatusInternalServerError)
+			}
+			return nil
+		})
+
+		var commandsCount int64
+		g2.Go(func() error {
+			var err error
+			if commandsCount, err = a.Srv().Store.Command().AnalyticsCommandCount(teamID); err != nil {
+				return model.NewAppError("GetAnalytics", "app.analytics.getanalytics.internal_error", nil, err.Error(), http.StatusInternalServerError)
+			}
+			return nil
+		})
+
+		var sessionsCount int64
+		g2.Go(func() error {
+			var err error
+			if sessionsCount, err = a.Srv().Store.Session().AnalyticsSessionCount(); err != nil {
+				return model.NewAppError("GetAnalytics", "app.session.analytics_session_count.app_error", nil, err.Error(), http.StatusInternalServerError)
+			}
+			return nil
+		})
+
+		var filesCount int64
+		var hashtagsCount int64
+		if !skipIntensiveQueries {
+			g2.Go(func() error {
+				var err error
+				if filesCount, err = a.Srv().Store.Post().AnalyticsPostCount(teamID, true, false); err != nil {
+					return model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, err.Error(), http.StatusInternalServerError)
+				}
+				return nil
+			})
+
+			g2.Go(func() error {
+				var err error
+				if hashtagsCount, err = a.Srv().Store.Post().AnalyticsPostCount(teamID, false, true); err != nil {
+					return model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, err.Error(), http.StatusInternalServerError)
+				}
+				return nil
+			})
 		}
 
-		if hashtagChan == nil {
+		if err := g2.Wait(); err != nil {
+			return nil, err.(*model.AppError)
+		}
+
+		if skipIntensiveQueries {
+			rows[0].Value = -1
 			rows[1].Value = -1
 		} else {
-			r := <-hashtagChan
+			rows[0].Value = float64(filesCount)
-			if r.NErr != nil {
+			rows[1].Value = float64(hashtagsCount)
-				return nil, model.NewAppError("GetAnalytics", "app.post.analytics_posts_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-			}
-			rows[1].Value = float64(r.Data.(int64))
 		}
 
-		r := <-iHookChan
+		rows[2].Value = float64(incomingWebhookCount)
-		if r.NErr != nil {
+		rows[3].Value = float64(outgoingWebhookCount)
-			return nil, model.NewAppError("GetAnalytics", "app.webhooks.analytics_incoming_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
+		rows[4].Value = float64(commandsCount)
-		}
+		rows[5].Value = float64(sessionsCount)
-		rows[2].Value = float64(r.Data.(int64))
-
-		r = <-oHookChan
-		if r.NErr != nil {
-			return nil, model.NewAppError("GetAnalytics", "app.webhooks.analytics_outgoing_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[3].Value = float64(r.Data.(int64))
-
-		r = <-commandChan
-		if r.NErr != nil {
-			return nil, model.NewAppError("GetAnalytics", "app.analytics.getanalytics.internal_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[4].Value = float64(r.Data.(int64))
-
-		r = <-sessionChan
-		if r.NErr != nil {
-			return nil, model.NewAppError("GetAnalytics", "app.session.analytics_session_count.app_error", nil, r.NErr.Error(), http.StatusInternalServerError)
-		}
-		rows[5].Value = float64(r.Data.(int64))
 
 		return rows, nil
 	}