// newStreamer returns sync or async streamer based on the configuration
// of the server and the session, sync streamer sends the events
// directly to the auth server and blocks if the events can not be received,
// async streamer buffers the events to disk and uploads the events later
newStreamer returns sync or async streamer based on the configuration
of the server and the session, sync streamer sends the events
directly to the auth server and blocks if the events can not be received,
async streamer buffers the events to disk and uploads the events later
// newStreamer returns sync or async streamer based on the configuration // of the server and the session sync streamer sends the events // directly to the auth server and blocks if the events can not be received // async streamer buffers the events to disk and uploads the events later
func (s *Server) newStreamer(ctx context.Context, sessionID string, clusterConfig services.ClusterConfig) (libevents.Streamer, error) {
	mode := clusterConfig.GetSessionRecording()
	if services.IsRecordSync(mode) {
		s.log.Debugf("Using sync streamer for session %v.", sessionID)
		return s.cfg.AuthClient, nil
	}
	s.log.Debugf("Using async streamer for session %v.", sessionID)
	uploadDir := filepath.Join(
		s.cfg.DataDir, teleport.LogsDir, teleport.ComponentUpload,
		libevents.StreamingLogsDir, defaults.Namespace)
	_, err := utils.StatDir(uploadDir)
	if err != nil && !trace.IsNotFound(err) {
		return nil, trace.Wrap(err)
	}
	if trace.IsNotFound(err) {
		s.log.Debugf("Creating upload dir %v.", uploadDir)
		if err := os.MkdirAll(uploadDir, 0755); err != nil {
			return nil, trace.Wrap(err)
		}
	}
	fileStreamer, err := filesessions.NewStreamer(uploadDir)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return libevents.NewTeeStreamer(fileStreamer, s.cfg.StreamEmitter), nil
}
func ( s * Server ) newStreamer ( ctx context . Context , sessionID string , clusterConfig services . ClusterConfig ) ( libevents . Streamer , error ) { mode := clusterConfig . GetSessionRecording ( ) if services . IsRecordSync ( mode ) { s . log . Debugf ( " " , sessionID ) return s . cfg . AuthClient , nil } s . log . Debugf ( " " , sessionID ) uploadDir := filepath . Join ( s . cfg . DataDir , teleport . LogsDir , teleport . ComponentUpload , libevents . StreamingLogsDir , defaults . Namespace ) _ , err := utils . StatDir ( uploadDir ) if err != nil && ! trace . IsNotFound ( err ) { return nil , trace . Wrap ( err ) } if trace . IsNotFound ( err ) { s . log . Debugf ( " " , uploadDir ) if err := os . MkdirAll ( uploadDir , 0755 ) ; err != nil { return nil , trace . Wrap ( err ) } } fileStreamer , err := filesessions . NewStreamer ( uploadDir ) if err != nil { return nil , trace . Wrap ( err ) } return libevents . NewTeeStreamer ( fileStreamer , s . cfg . StreamEmitter ) , nil }
newStreamer returns sync or async streamer based on the configuration
of the server and the session, sync streamer sends the events
directly to the auth server and blocks if the events can not be received,
async streamer buffers the events to disk and uploads the events later
newStreamer returns sync or async streamer based on the configuration of the server and the session sync streamer sends the events directly to the auth server and blocks if the events can not be received async streamer buffers the events to disk and uploads the events later
// Make sure the upload dir exists, otherwise file streamer will fail.